Source code for argopy.utilities

#!/bin/env python
# -*coding: UTF-8 -*-
#
# Disclaimer:
# Functions get_sys_info, netcdf_and_hdf5_versions and show_versions are from:
#   xarray/util/print_versions.py
#

import os
import sys
import warnings
import urllib
import json
import collections
from collections import UserList
import copy
from functools import reduce, wraps
from packaging import version
import logging
from abc import ABC, abstractmethod
from urllib.parse import urlparse
from typing import Union
import inspect
import pathlib
import importlib
import locale
import platform
import struct
import subprocess  # nosec B404 only used without user inputs
import contextlib
from fsspec.core import split_protocol
import fsspec

import argopy
import xarray as xr
import pandas as pd
import numpy as np
from scipy import interpolate

import pickle  # nosec B403 only used with internal files/assets
import pkg_resources
import shutil

import threading
from socket import gaierror

import time
import setuptools  # noqa: F401

from .options import OPTIONS
from .errors import (
    FtpPathError,
    InvalidFetcher,
    InvalidFetcherAccessPoint,
    InvalidOption,
    InvalidDatasetStructure,
    FileSystemHasNoCache,
    DataNotFound,
)

try:
    collectionsAbc = collections.abc
except AttributeError:
    collectionsAbc = collections

try:
    importlib.import_module('matplotlib')  # noqa: E402
    from matplotlib.colors import to_hex
except ImportError:
    pass

path2pkl = pkg_resources.resource_filename("argopy", "assets/")

log = logging.getLogger("argopy.utilities")


[docs]def clear_cache(fs=None): """ Delete argopy cache folder content """ if os.path.exists(OPTIONS["cachedir"]): # shutil.rmtree(OPTIONS["cachedir"]) for filename in os.listdir(OPTIONS["cachedir"]): file_path = os.path.join(OPTIONS["cachedir"], filename) try: if os.path.isfile(file_path) or os.path.islink(file_path): os.unlink(file_path) elif os.path.isdir(file_path): shutil.rmtree(file_path) except Exception as e: print("Failed to delete %s. Reason: %s" % (file_path, e)) if fs: fs.clear_cache()
def lscache(cache_path: str = "", prt=True): """ Decode and list cache folder content Parameters ---------- cache_path: str prt: bool, default=True Return a printable string or a :class:`pandas.DataFrame` Returns ------- str or :class:`pandas.DataFrame` """ from datetime import datetime import math summary = [] cache_path = OPTIONS['cachedir'] if cache_path == '' else cache_path apath = os.path.abspath(cache_path) log.debug("Listing cache content at: %s" % cache_path) def convert_size(size_bytes): if size_bytes == 0: return "0B" size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB") i = int(math.floor(math.log(size_bytes, 1024))) p = math.pow(1024, i) s = round(size_bytes / p, 2) return "%s %s" % (s, size_name[i]) cached_files = [] fn = os.path.join(apath, "cache") if os.path.exists(fn): with open(fn, "rb") as f: loaded_cached_files = pickle.load(f) # nosec B301 because files controlled internally for c in loaded_cached_files.values(): if isinstance(c["blocks"], list): c["blocks"] = set(c["blocks"]) cached_files.append(loaded_cached_files) else: raise FileSystemHasNoCache("No fsspec cache system at: %s" % apath) cached_files = cached_files or [{}] cached_files = cached_files[-1] N_FILES = len(cached_files) TOTAL_SIZE = 0 for cfile in cached_files: path = os.path.join(apath, cached_files[cfile]['fn']) TOTAL_SIZE += os.path.getsize(path) summary.append("%s %s" % ("=" * 20, "%i files in fsspec cache folder (%s)" % (N_FILES, convert_size(TOTAL_SIZE)))) summary.append("lscache %s" % os.path.sep.join([apath, ""])) summary.append("=" * 20) listing = {'fn': [], 'size': [], 'time': [], 'original': [], 'uid': [], 'blocks': []} for cfile in cached_files: summary.append("- %s" % cached_files[cfile]['fn']) listing['fn'].append(cached_files[cfile]['fn']) path = os.path.join(cache_path, cached_files[cfile]['fn']) summary.append("\t%8s: %s" % ('SIZE', convert_size(os.path.getsize(path)))) listing['size'].append(os.path.getsize(path)) key = 'time' ts = cached_files[cfile][key] tsf = pd.to_datetime(datetime.fromtimestamp(ts)).strftime("%c") summary.append("\t%8s: %s (%s)" % (key, tsf, ts)) listing['time'].append(pd.to_datetime(datetime.fromtimestamp(ts))) if version.parse(fsspec.__version__) > version.parse("0.8.7"): key = 'original' summary.append("\t%8s: %s" % (key, cached_files[cfile][key])) listing[key].append(cached_files[cfile][key]) key = 'uid' summary.append("\t%8s: %s" % (key, cached_files[cfile][key])) listing[key].append(cached_files[cfile][key]) key = 'blocks' summary.append("\t%8s: %s" % (key, cached_files[cfile][key])) listing[key].append(cached_files[cfile][key]) summary.append("=" * 20) summary = "\n".join(summary) if prt: # Return string to be printed: return summary else: # Return dataframe listing: # log.debug(summary) return pd.DataFrame(listing) def load_dict(ptype): if ptype == "profilers": with open(os.path.join(path2pkl, "dict_profilers.pickle"), "rb") as f: loaded_dict = pickle.load(f) # nosec B301 because files controlled internally return loaded_dict elif ptype == "institutions": with open(os.path.join(path2pkl, "dict_institutions.pickle"), "rb") as f: loaded_dict = pickle.load(f) # nosec B301 because files controlled internally return loaded_dict else: raise ValueError("Invalid dictionary pickle file") def mapp_dict(Adictionnary, Avalue): if Avalue not in Adictionnary: return "Unknown" else: return Adictionnary[Avalue]
[docs]def list_available_data_src(): """ List all available data sources """ sources = {} try: from .data_fetchers import erddap_data as Erddap_Fetchers # Ensure we're loading the erddap data fetcher with the current options: Erddap_Fetchers.api_server_check = Erddap_Fetchers.api_server_check.replace(Erddap_Fetchers.api_server, OPTIONS['erddap']) Erddap_Fetchers.api_server = OPTIONS['erddap'] sources["erddap"] = Erddap_Fetchers except Exception: warnings.warn( "An error occurred while loading the ERDDAP data fetcher, " "it will not be available !\n%s\n%s" % (sys.exc_info()[0], sys.exc_info()[1]) ) pass try: from .data_fetchers import argovis_data as ArgoVis_Fetchers sources["argovis"] = ArgoVis_Fetchers except Exception: warnings.warn( "An error occurred while loading the ArgoVis data fetcher, " "it will not be available !\n%s\n%s" % (sys.exc_info()[0], sys.exc_info()[1]) ) pass try: from .data_fetchers import gdacftp_data as GDAC_Fetchers # Ensure we're loading the gdac data fetcher with the current options: GDAC_Fetchers.api_server_check = OPTIONS['ftp'] GDAC_Fetchers.api_server = OPTIONS['ftp'] sources["gdac"] = GDAC_Fetchers except Exception: warnings.warn( "An error occurred while loading the GDAC data fetcher, " "it will not be available !\n%s\n%s" % (sys.exc_info()[0], sys.exc_info()[1]) ) pass # return dict(sorted(sources.items())) return sources
[docs]def list_available_index_src(): """ List all available index sources """ sources = {} try: from .data_fetchers import erddap_index as Erddap_Fetchers # Ensure we're loading the erddap data fetcher with the current options: Erddap_Fetchers.api_server_check = Erddap_Fetchers.api_server_check.replace(Erddap_Fetchers.api_server, OPTIONS['erddap']) Erddap_Fetchers.api_server = OPTIONS['erddap'] sources["erddap"] = Erddap_Fetchers except Exception: warnings.warn( "An error occurred while loading the ERDDAP index fetcher, " "it will not be available !\n%s\n%s" % (sys.exc_info()[0], sys.exc_info()[1]) ) pass try: from .data_fetchers import gdacftp_index as GDAC_Fetchers # Ensure we're loading the gdac data fetcher with the current options: GDAC_Fetchers.api_server_check = OPTIONS['ftp'] GDAC_Fetchers.api_server = OPTIONS['ftp'] sources["gdac"] = GDAC_Fetchers except Exception: warnings.warn( "An error occurred while loading the GDAC index fetcher, " "it will not be available !\n%s\n%s" % (sys.exc_info()[0], sys.exc_info()[1]) ) pass return sources
[docs]def list_standard_variables(): """ List of variables for standard users """ return [ "DATA_MODE", "LATITUDE", "LONGITUDE", "POSITION_QC", "DIRECTION", "PLATFORM_NUMBER", "CYCLE_NUMBER", "PRES", "TEMP", "PSAL", "PRES_QC", "TEMP_QC", "PSAL_QC", "PRES_ADJUSTED", "TEMP_ADJUSTED", "PSAL_ADJUSTED", "PRES_ADJUSTED_QC", "TEMP_ADJUSTED_QC", "PSAL_ADJUSTED_QC", "PRES_ADJUSTED_ERROR", "TEMP_ADJUSTED_ERROR", "PSAL_ADJUSTED_ERROR", "PRES_ERROR", # can be created from PRES_ADJUSTED_ERROR after a filter_data_mode "TEMP_ERROR", "PSAL_ERROR", "JULD", "JULD_QC", "TIME", "TIME_QC", # "CONFIG_MISSION_NUMBER", ]
[docs]def list_multiprofile_file_variables(): """ List of variables in a netcdf multiprofile file. This is for files created by GDAC under <DAC>/<WMO>/<WMO>_prof.nc """ return [ "CONFIG_MISSION_NUMBER", "CYCLE_NUMBER", "DATA_CENTRE", "DATA_MODE", "DATA_STATE_INDICATOR", "DATA_TYPE", "DATE_CREATION", "DATE_UPDATE", "DC_REFERENCE", "DIRECTION", "FIRMWARE_VERSION", "FLOAT_SERIAL_NO", "FORMAT_VERSION", "HANDBOOK_VERSION", "HISTORY_ACTION", "HISTORY_DATE", "HISTORY_INSTITUTION", "HISTORY_PARAMETER", "HISTORY_PREVIOUS_VALUE", "HISTORY_QCTEST", "HISTORY_REFERENCE", "HISTORY_SOFTWARE", "HISTORY_SOFTWARE_RELEASE", "HISTORY_START_PRES", "HISTORY_STEP", "HISTORY_STOP_PRES", "JULD", "JULD_LOCATION", "JULD_QC", "LATITUDE", "LONGITUDE", "PARAMETER", "PI_NAME", "PLATFORM_NUMBER", "PLATFORM_TYPE", "POSITIONING_SYSTEM", "POSITION_QC", "PRES", "PRES_ADJUSTED", "PRES_ADJUSTED_ERROR", "PRES_ADJUSTED_QC", "PRES_QC", "PROFILE_PRES_QC", "PROFILE_PSAL_QC", "PROFILE_TEMP_QC", "PROJECT_NAME", "PSAL", "PSAL_ADJUSTED", "PSAL_ADJUSTED_ERROR", "PSAL_ADJUSTED_QC", "PSAL_QC", "REFERENCE_DATE_TIME", "SCIENTIFIC_CALIB_COEFFICIENT", "SCIENTIFIC_CALIB_COMMENT", "SCIENTIFIC_CALIB_DATE", "SCIENTIFIC_CALIB_EQUATION", "STATION_PARAMETERS", "TEMP", "TEMP_ADJUSTED", "TEMP_ADJUSTED_ERROR", "TEMP_ADJUSTED_QC", "TEMP_QC", "VERTICAL_SAMPLING_SCHEME", "WMO_INST_TYPE", ]
def get_sys_info(): "Returns system information as a dict" blob = [] # get full commit hash commit = None if os.path.isdir(".git") and os.path.isdir("argopy"): try: pipe = subprocess.Popen( # nosec No user provided input to control here 'git log --format="%H" -n 1'.split(" "), stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) so, serr = pipe.communicate() except Exception: pass else: if pipe.returncode == 0: commit = so try: commit = so.decode("utf-8") except ValueError: pass commit = commit.strip().strip('"') blob.append(("commit", commit)) try: (sysname, nodename, release, version_, machine, processor) = platform.uname() blob.extend( [ ("python", sys.version), ("python-bits", struct.calcsize("P") * 8), ("OS", "%s" % (sysname)), ("OS-release", "%s" % (release)), ("machine", "%s" % (machine)), ("processor", "%s" % (processor)), ("byteorder", "%s" % sys.byteorder), ("LC_ALL", "%s" % os.environ.get("LC_ALL", "None")), ("LANG", "%s" % os.environ.get("LANG", "None")), ("LOCALE", "%s.%s" % locale.getlocale()), ] ) except Exception: pass return blob def netcdf_and_hdf5_versions(): libhdf5_version = None libnetcdf_version = None try: import netCDF4 libhdf5_version = netCDF4.__hdf5libversion__ libnetcdf_version = netCDF4.__netcdf4libversion__ except ImportError: try: import h5py libhdf5_version = h5py.version.hdf5_version except ImportError: pass return [("libhdf5", libhdf5_version), ("libnetcdf", libnetcdf_version)]
[docs]def show_versions(file=sys.stdout, conda=False): # noqa: C901 """ Print the versions of argopy and its dependencies Parameters ---------- file : file-like, optional print to the given file-like object. Defaults to sys.stdout. conda: bool, optional format versions to be copy/pasted on a conda environment file (default, False) """ sys_info = get_sys_info() try: sys_info.extend(netcdf_and_hdf5_versions()) except Exception as e: print(f"Error collecting netcdf / hdf5 version: {e}") DEPS = { 'core': sorted([ ("argopy", lambda mod: mod.__version__), ("xarray", lambda mod: mod.__version__), ("scipy", lambda mod: mod.__version__), ("netCDF4", lambda mod: mod.__version__), ("erddapy", lambda mod: mod.__version__), # This could go away from requirements ? ("fsspec", lambda mod: mod.__version__), ("aiohttp", lambda mod: mod.__version__), ("packaging", lambda mod: mod.__version__), # will come with xarray, Using 'version' to make API compatible with several fsspec releases ("toolz", lambda mod: mod.__version__), ("requests", lambda mod: mod.__version__), ]), 'ext.util': sorted([ ("gsw", lambda mod: mod.__version__), # Used by xarray accessor to compute new variables ("tqdm", lambda mod: mod.__version__), ("zarr", lambda mod: mod.__version__), ]), 'ext.perf': sorted([ ("dask", lambda mod: mod.__version__), ("distributed", lambda mod: mod.__version__), ("pyarrow", lambda mod: mod.__version__), ]), 'ext.plot': sorted([ ("matplotlib", lambda mod: mod.__version__), ("cartopy", lambda mod: mod.__version__), ("seaborn", lambda mod: mod.__version__), ("IPython", lambda mod: mod.__version__), ("ipywidgets", lambda mod: mod.__version__), ("ipykernel", lambda mod: mod.__version__), ]), 'dev': sorted([ ("bottleneck", lambda mod: mod.__version__), ("cftime", lambda mod: mod.__version__), ("cfgrib", lambda mod: mod.__version__), ("conda", lambda mod: mod.__version__), ("nc_time_axis", lambda mod: mod.__version__), ("numpy", lambda mod: mod.__version__), # will come with xarray and pandas ("pandas", lambda mod: mod.__version__), # will come with xarray ("pip", lambda mod: mod.__version__), ("black", lambda mod: mod.__version__), ("flake8", lambda mod: mod.__version__), ("pytest", lambda mod: mod.__version__), # will come with pandas ("pytest_env", lambda mod: mod.__version__), # will come with pandas ("pytest_cov", lambda mod: mod.__version__), # will come with pandas ("pytest_localftpserver", lambda mod: mod.__version__), # will come with pandas ("setuptools", lambda mod: mod.__version__), # Provides: pkg_resources ("sphinx", lambda mod: mod.__version__), ]), } DEPS_blob = {} for level in DEPS.keys(): deps = DEPS[level] deps_blob = list() for (modname, ver_f) in deps: try: if modname in sys.modules: mod = sys.modules[modname] else: mod = importlib.import_module(modname) except Exception: deps_blob.append((modname, '-')) else: try: ver = ver_f(mod) deps_blob.append((modname, ver)) except Exception: deps_blob.append((modname, "installed")) DEPS_blob[level] = deps_blob print("\nSYSTEM", file=file) print("------", file=file) for k, stat in sys_info: print(f"{k}: {stat}", file=file) for level in DEPS_blob: if conda: print("\n# %s:" % level.upper(), file=file) else: title = "INSTALLED VERSIONS: %s" % level.upper() print("\n%s" % title, file=file) print("-"*len(title), file=file) deps_blob = DEPS_blob[level] for k, stat in deps_blob: if conda: if k != 'argopy': kf = k.replace("_", "-") comment = ' ' if stat != '-' else '# ' print(f"{comment} - {kf} = {stat}", file=file) # Format like a conda env line, useful to update ci/requirements else: print("{:<12}: {:<12}".format(k, stat), file=file)
[docs]def show_options(file=sys.stdout): # noqa: C901 """ Print options of argopy Parameters ---------- file : file-like, optional print to the given file-like object. Defaults to sys.stdout. """ print("\nARGOPY OPTIONS", file=file) print("--------------", file=file) opts = copy.deepcopy(OPTIONS) opts = dict(sorted(opts.items())) for k, v in opts.items(): print(f"{k}: {v}", file=file)
def check_gdac_path(path, errors='ignore'): # noqa: C901 """ Check if a path has the expected GDAC ftp structure Expected GDAC ftp structure:: . └── dac β”œβ”€β”€ aoml β”œβ”€β”€ ... β”œβ”€β”€ coriolis β”œβ”€β”€ ... β”œβ”€β”€ meds └── nmdis This check will return True if at least one DAC sub-folder is found under path/dac/<dac_name> Examples:: >>> check_gdac_path("https://data-argo.ifremer.fr") # True >>> check_gdac_path("ftp://ftp.ifremer.fr/ifremer/argo") # True >>> check_gdac_path("ftp://usgodae.org/pub/outgoing/argo") # True >>> check_gdac_path("/home/ref-argo/gdac") # True >>> check_gdac_path("https://www.ifremer.fr") # False >>> check_gdac_path("ftp://usgodae.org/pub/outgoing") # False Parameters ---------- path: str Path name to check, including access protocol errors: str "ignore" or "raise" (or "warn") Returns ------- checked: boolean True if at least one DAC folder is found under path/dac/<dac_name> False otherwise """ # Create a file system for this path if split_protocol(path)[0] is None: fs = fsspec.filesystem('file') elif 'https' in split_protocol(path)[0]: fs = fsspec.filesystem('http') elif 'ftp' in split_protocol(path)[0]: try: host = split_protocol(path)[-1].split('/')[0] fs = fsspec.filesystem('ftp', host=host) except gaierror: if errors == 'raise': raise FtpPathError("Can't get address info (GAIerror) on '%s'" % host) elif errors == "warn": warnings.warn("Can't get address info (GAIerror) on '%s'" % host) return False else: return False else: raise FtpPathError("Unknown protocol for an Argo GDAC host: %s" % split_protocol(path)[0]) # dacs = [ # "aoml", # "bodc", # "coriolis", # "csio", # "csiro", # "incois", # "jma", # "kma", # "kordi", # "meds", # "nmdis", # ] # Case 1: check1 = ( fs.exists(path) and fs.exists(fs.sep.join([path, "dac"])) # and np.any([fs.exists(fs.sep.join([path, "dac", dac])) for dac in dacs]) # Take too much time on http/ftp GDAC server ) if check1: return True elif errors == "raise": raise FtpPathError("This path is not GDAC compliant (no `dac` folder with legitimate sub-folder):\n%s" % path) elif errors == "warn": warnings.warn("This path is not GDAC compliant:\n%s" % path) return False else: return False
[docs]def isconnected(host: str = "https://www.ifremer.fr", maxtry: int = 10): """Check if an URL is alive Parameters ---------- host: str URL to use, 'https://www.ifremer.fr' by default maxtry: int, default: 10 Maximum number of host connections to try before Returns ------- bool """ # log.debug("isconnected: %s" % host) if split_protocol(host)[0] in ["http", "https", "ftp", "sftp"]: it = 0 while it < maxtry: try: # log.debug("Checking if %s is connected ..." % host) urllib.request.urlopen(host, timeout=1) # nosec B310 because host protocol already checked result, it = True, maxtry except Exception: result, it = False, it+1 return result else: return os.path.exists(host)
[docs]def urlhaskeyword(url: str = "", keyword: str = '', maxtry: int = 10): """ Check if a keyword is in the content of a URL Parameters ---------- url: str keyword: str maxtry: int, default: 10 Maximum number of host connections to try before returning False Returns ------- bool """ it = 0 while it < maxtry: try: with fsspec.open(url) as f: data = f.read() result = keyword in str(data) it = maxtry except Exception: result, it = False, it + 1 return result
[docs]def isalive(api_server_check: Union[str, dict] = "") -> bool: """Check if an API is alive or not 2 methods are available: - URL Ping - keyword Check Parameters ---------- api_server_check Url string or dictionary with [``url``, ``keyword``] keys. - For a string, uses: :class:`argopy.utilities.isconnected` - For a dictionary, uses: :class:`argopy.utilities.urlhaskeyword` Returns ------- bool """ # log.debug("isalive: %s" % api_server_check) if isinstance(api_server_check, dict): return urlhaskeyword(url=api_server_check['url'], keyword=api_server_check['keyword']) else: return isconnected(api_server_check)
[docs]def isAPIconnected(src="erddap", data=True): """ Check if a source API is alive or not The API is connected when it has a live URL or valid folder path. Parameters ---------- src: str The data or index source name, 'erddap' default data: bool If True check the data fetcher (default), if False, check the index fetcher Returns ------- bool """ if data: list_src = list_available_data_src() else: list_src = list_available_index_src() if src in list_src and getattr(list_src[src], "api_server_check", None): return isalive(list_src[src].api_server_check) else: raise InvalidFetcher
def erddap_ds_exists( ds: Union[list, str] = "ArgoFloats", erddap: str = None, maxtry: int = 2 ) -> bool: """ Check if a dataset exists on a remote erddap server Parameter --------- ds: str, default='ArgoFloats' Name of the erddap dataset to check erddap: str, default=OPTIONS['erddap'] Url of the erddap server maxtry: int, default: 2 Maximum number of host connections to try Return ------ bool """ if erddap is None: erddap = OPTIONS['erddap'] # log.debug("from erddap_ds_exists: %s" % erddap) from .stores import httpstore if isconnected(erddap, maxtry=maxtry): with httpstore(timeout=OPTIONS['api_timeout']).open("".join([erddap, "/info/index.json"])) as of: erddap_index = json.load(of) if is_list_of_strings(ds): return [this_ds in [row[-1] for row in erddap_index["table"]["rows"]] for this_ds in ds] else: return ds in [row[-1] for row in erddap_index["table"]["rows"]] else: log.debug("Cannot reach erddap server: %s" % erddap) warnings.warn("Return False because we cannot reach the erddap server %s" % erddap) return False def badge(label="label", message="message", color="green", insert=False): """ Return or insert shield.io badge image Use the shields.io service to create a badge image https://img.shields.io/static/v1?label=<LABEL>&message=<MESSAGE>&color=<COLOR> Parameters ---------- label: str Left side badge text message: str Right side badge text color: str Right side background color insert: bool Return url to badge image (False, default) or directly insert the image with HTML (True) Returns ------- str or IPython.display.Image """ from IPython.display import Image url = ( "https://img.shields.io/static/v1?style=flat-square&label={}&message={}&color={}" ).format img = url(urllib.parse.quote(label), urllib.parse.quote(message), color) if not insert: return img else: return Image(url=img) class fetch_status: """Fetch and report web API status""" def __init__(self, **kwargs): if "stdout" in kwargs or "insert" in kwargs: warnings.warn("'fetch_status' signature has changed") pass def fetch(self): results = {} list_src = list_available_data_src() for api, mod in list_src.items(): if getattr(mod, "api_server_check", None): status = isAPIconnected(api) message = "ok" if status else "offline" results[api] = {"value": status, "message": message} return results @property def text(self): results = self.fetch() rows = [] for api in sorted(results.keys()): rows.append("src %s is: %s" % (api, results[api]["message"])) txt = " | ".join(rows) return txt def __repr__(self): return self.text @property def html(self): results = self.fetch() fs = 12 def td_msg(bgcolor, txtcolor, txt): style = "background-color:%s;" % to_hex(bgcolor, keep_alpha=True) style+= "border-width:0px;" style+= "padding: 2px 5px 2px 5px;" style+= "text-align:left;" style+= "color:%s" % to_hex(txtcolor, keep_alpha=True) return "<td style='%s'>%s</td>" % (style, str(txt)) td_empty = "<td style='border-width:0px;padding: 2px 5px 2px 5px;text-align:left'>&nbsp;</td>" html = [] html.append("<table style='border-collapse:collapse;border-spacing:0;font-size:%ipx'>" % fs) html.append("<tbody><tr>") cols = [] for api in sorted(results.keys()): color = "yellowgreen" if results[api]["value"] else "darkorange" cols.append(td_msg('dimgray', 'w', "src %s is" % api)) cols.append(td_msg(color, 'w', results[api]["message"])) cols.append(td_empty) html.append("\n".join(cols)) html.append("</tr></tbody>") html.append("</table>") html = "\n".join(html) return html def _repr_html_(self): return self.html
[docs]class monitor_status: """ Monitor data source status with a refresh rate """
[docs] def __init__(self, refresh=60): self.refresh_rate = refresh if self.runner == 'notebook': import ipywidgets as widgets self.text = widgets.HTML( value=self.content, placeholder="", description="", ) self.start()
def __repr__(self): if self.runner != 'notebook': return self.content else: return "" @property def runner(self) -> str: try: shell = get_ipython().__class__.__name__ if shell == 'ZMQInteractiveShell': return 'notebook' # Jupyter notebook or qtconsole elif shell == 'TerminalInteractiveShell': return 'terminal' # Terminal running IPython else: return False # Other type (?) except NameError: return 'standard' # Probably standard Python interpreter @property def content(self): if self.runner == 'notebook': return fetch_status().html else: return fetch_status().text def work(self): while True: time.sleep(self.refresh_rate) self.text.value = self.content def start(self): from IPython.display import display thread = threading.Thread(target=self.work) display(self.text) thread.start()
# # From xarrayutils : https://github.com/jbusecke/xarrayutils/blob/master/xarrayutils/vertical_coordinates.py # Β Direct integration of those 2 functions to minimize dependencies and possibility of tuning them to our needs #
[docs]def linear_interpolation_remap( z, data, z_regridded, z_dim=None, z_regridded_dim="regridded", output_dim="remapped" ): # interpolation called in xarray ufunc def _regular_interp(x, y, target_values): # remove all nans from input x and y idx = np.logical_or(np.isnan(x), np.isnan(y)) x = x[~idx] y = y[~idx] # Need at least 5 points in the profile to interpolate, otherwise, return NaNs if len(y) < 5: interpolated = np.empty(len(target_values)) interpolated[:] = np.nan else: # replace nans in target_values with out of bound Values (just in case) target_values = np.where( ~np.isnan(target_values), target_values, np.nanmax(x) + 1 ) # Interpolate with fill value parameter to extend min pressure toward 0 interpolated = interpolate.interp1d( x, y, bounds_error=False, fill_value=(y[0], y[-1]) )(target_values) return interpolated # infer dim from input if z_dim is None: if len(z.dims) != 1: raise RuntimeError("if z_dim is not specified, x must be a 1D array.") dim = z.dims[0] else: dim = z_dim # if dataset is passed drop all data_vars that dont contain dim if isinstance(data, xr.Dataset): raise ValueError("Dataset input is not supported yet") # TODO: for a dataset input just apply the function for each appropriate array if version.parse(xr.__version__) > version.parse("0.15.0"): kwargs = dict( input_core_dims=[[dim], [dim], [z_regridded_dim]], output_core_dims=[[output_dim]], vectorize=True, dask="parallelized", output_dtypes=[data.dtype], dask_gufunc_kwargs={ "output_sizes": {output_dim: len(z_regridded[z_regridded_dim])} }, ) else: kwargs = dict( input_core_dims=[[dim], [dim], [z_regridded_dim]], output_core_dims=[[output_dim]], vectorize=True, dask="parallelized", output_dtypes=[data.dtype], output_sizes={output_dim: len(z_regridded[z_regridded_dim])}, ) remapped = xr.apply_ufunc(_regular_interp, z, data, z_regridded, **kwargs) remapped.coords[output_dim] = z_regridded.rename( {z_regridded_dim: output_dim} ).coords[output_dim] return remapped
[docs]class Chunker: """ To chunk fetcher requests """ # Default maximum chunks size for all possible request parameters default_chunksize = { "box": { "lon": 20, # degree "lat": 20, # degree "dpt": 500, # meters/db "time": 3 * 30, }, # Days "wmo": {"wmo": 5, "cyc": 100}, # Nb of floats } # Nb of cycles
[docs] def __init__(self, request: dict, chunks: str = "auto", chunksize: dict = {}): """ Create a request Chunker Allow to easily split an access point request into chunks Parameters ---------- request: dict Access point request to be chunked. One of the following: - {'box': [lon_min, lon_max, lat_min, lat_max, dpt_min, dpt_max, time_min, time_max]} - {'box': [lon_min, lon_max, lat_min, lat_max, dpt_min, dpt_max]} - {'wmo': [wmo1, wmo2, ...], 'cyc': [0,1, ...]} chunks: 'auto' or dict Dictionary with request access point as keys and number of chunks to create as values. Eg: {'wmo':10} will create a maximum of 10 chunks along WMOs. chunksize: dict, optional Dictionary with request access point as keys and chunk size as values (used as maximum values in 'auto' chunking). Eg: {'wmo': 5} will create chunks with as many as 5 WMOs each. """ self.request = request if "box" in self.request: is_box(self.request["box"]) if len(self.request["box"]) == 8: self.this_chunker = self._chunker_box4d elif len(self.request["box"]) == 6: self.this_chunker = self._chunker_box3d elif "wmo" in self.request: self.this_chunker = self._chunker_wmo else: raise InvalidFetcherAccessPoint( "'%s' not valid access point" % ",".join(self.request.keys()) ) default = self.default_chunksize[[k for k in self.request.keys()][0]] if len(chunksize) == 0: # chunksize = {} chunksize = default if not isinstance(chunksize, collectionsAbc.Mapping): raise ValueError("chunksize must be mappable") else: # merge with default: chunksize = {**default, **chunksize} self.chunksize = collections.OrderedDict(sorted(chunksize.items())) default = {k: "auto" for k in self.chunksize.keys()} if chunks == "auto": # auto for all chunks = default elif len(chunks) == 0: # chunks = {}, i.e. chunk=1 for all chunks = {k: 1 for k in self.request} if not isinstance(chunks, collectionsAbc.Mapping): raise ValueError("chunks must be 'auto' or mappable") chunks = {**default, **chunks} self.chunks = collections.OrderedDict(sorted(chunks.items()))
def _split(self, lst, n=1): """Yield successive n-sized chunks from lst""" for i in range(0, len(lst), n): yield lst[i: i + n] def _split_list_bychunknb(self, lst, n=1): """Split list in n-imposed chunks of similar size The last chunk may contain more or less element than the others, depending on the size of the list. """ res = [] siz = int(np.floor_divide(len(lst), n)) for i in self._split(lst, siz): res.append(i) if len(res) > n: res[n-1::] = [reduce(lambda i, j: i + j, res[n-1::])] return res def _split_list_bychunksize(self, lst, max_size=1): """Split list in chunks of imposed size The last chunk may contain more or less element than the others, depending on the size of the list. """ res = [] for i in self._split(lst, max_size): res.append(i) return res def _split_box(self, large_box, n=1, d="x"): # noqa: C901 """Split a box domain in one direction in n-imposed equal chunks """ if d == "x": i_left, i_right = 0, 1 if d == "y": i_left, i_right = 2, 3 if d == "z": i_left, i_right = 4, 5 if d == "t": i_left, i_right = 6, 7 if n == 1: return [large_box] boxes = [] if d in ["x", "y", "z"]: n += 1 # Required because we split in linspace bins = np.linspace(large_box[i_left], large_box[i_right], n) for ii, left in enumerate(bins): if ii < len(bins) - 1: right = bins[ii + 1] this_box = large_box.copy() this_box[i_left] = left this_box[i_right] = right boxes.append(this_box) elif "t" in d: dates = pd.to_datetime(large_box[i_left: i_right + 1]) date_bounds = [ d.strftime("%Y%m%d%H%M%S") for d in pd.date_range(dates[0], dates[1], periods=n + 1) ] for i1, i2 in zip(np.arange(0, n), np.arange(1, n + 1)): left, right = date_bounds[i1], date_bounds[i2] this_box = large_box.copy() this_box[i_left] = left this_box[i_right] = right boxes.append(this_box) return boxes def _split_this_4Dbox(self, box, nx=1, ny=1, nz=1, nt=1): box_list = [] split_x = self._split_box(box, n=nx, d="x") for bx in split_x: split_y = self._split_box(bx, n=ny, d="y") for bxy in split_y: split_z = self._split_box(bxy, n=nz, d="z") for bxyz in split_z: split_t = self._split_box(bxyz, n=nt, d="t") for bxyzt in split_t: box_list.append(bxyzt) return box_list def _split_this_3Dbox(self, box, nx=1, ny=1, nz=1): box_list = [] split_x = self._split_box(box, n=nx, d="x") for bx in split_x: split_y = self._split_box(bx, n=ny, d="y") for bxy in split_y: split_z = self._split_box(bxy, n=nz, d="z") for bxyz in split_z: box_list.append(bxyz) return box_list def _chunker_box4d(self, request, chunks, chunks_maxsize): # noqa: C901 BOX = request["box"] n_chunks = chunks for axis, n in n_chunks.items(): if n == "auto": if axis == "lon": Lx = BOX[1] - BOX[0] if Lx > chunks_maxsize["lon"]: # Max box size in longitude n_chunks["lon"] = int( np.ceil(np.divide(Lx, chunks_maxsize["lon"])) ) else: n_chunks["lon"] = 1 if axis == "lat": Ly = BOX[3] - BOX[2] if Ly > chunks_maxsize["lat"]: # Max box size in latitude n_chunks["lat"] = int( np.ceil(np.divide(Ly, chunks_maxsize["lat"])) ) else: n_chunks["lat"] = 1 if axis == "dpt": Lz = BOX[5] - BOX[4] if Lz > chunks_maxsize["dpt"]: # Max box size in depth n_chunks["dpt"] = int( np.ceil(np.divide(Lz, chunks_maxsize["dpt"])) ) else: n_chunks["dpt"] = 1 if axis == "time": Lt = np.timedelta64( pd.to_datetime(BOX[7]) - pd.to_datetime(BOX[6]), "D" ) MaxLen = np.timedelta64(chunks_maxsize["time"], "D") if Lt > MaxLen: # Max box size in time n_chunks["time"] = int(np.ceil(np.divide(Lt, MaxLen))) else: n_chunks["time"] = 1 boxes = self._split_this_4Dbox( BOX, nx=n_chunks["lon"], ny=n_chunks["lat"], nz=n_chunks["dpt"], nt=n_chunks["time"], ) return {"chunks": sorted(n_chunks), "values": boxes} def _chunker_box3d(self, request, chunks, chunks_maxsize): BOX = request["box"] n_chunks = chunks for axis, n in n_chunks.items(): if n == "auto": if axis == "lon": Lx = BOX[1] - BOX[0] if Lx > chunks_maxsize["lon"]: # Max box size in longitude n_chunks["lon"] = int( np.floor_divide(Lx, chunks_maxsize["lon"]) ) else: n_chunks["lon"] = 1 if axis == "lat": Ly = BOX[3] - BOX[2] if Ly > chunks_maxsize["lat"]: # Max box size in latitude n_chunks["lat"] = int( np.floor_divide(Ly, chunks_maxsize["lat"]) ) else: n_chunks["lat"] = 1 if axis == "dpt": Lz = BOX[5] - BOX[4] if Lz > chunks_maxsize["dpt"]: # Max box size in depth n_chunks["dpt"] = int( np.floor_divide(Lz, chunks_maxsize["dpt"]) ) else: n_chunks["dpt"] = 1 # if axis == 'time': # Lt = np.timedelta64(pd.to_datetime(BOX[5]) - pd.to_datetime(BOX[4]), 'D') # MaxLen = np.timedelta64(chunks_maxsize['time'], 'D') # if Lt > MaxLen: # Max box size in time # n_chunks['time'] = int(np.floor_divide(Lt, MaxLen)) # else: # n_chunks['time'] = 1 boxes = self._split_this_3Dbox( BOX, nx=n_chunks["lon"], ny=n_chunks["lat"], nz=n_chunks["dpt"] ) return {"chunks": sorted(n_chunks), "values": boxes} def _chunker_wmo(self, request, chunks, chunks_maxsize): WMO = request["wmo"] n_chunks = chunks if n_chunks["wmo"] == "auto": wmo_grps = self._split_list_bychunksize(WMO, max_size=chunks_maxsize["wmo"]) else: n = np.min([n_chunks["wmo"], len(WMO)]) wmo_grps = self._split_list_bychunknb(WMO, n=n) n_chunks["wmo"] = len(wmo_grps) return {"chunks": sorted(n_chunks), "values": wmo_grps} def fit_transform(self): """ Chunk a fetcher request Returns ------- list """ self._results = self.this_chunker(self.request, self.chunks, self.chunksize) # self.chunks = self._results['chunks'] return self._results["values"]
[docs]def format_oneline(s, max_width=65): """ Return a string formatted for a line print """ if len(s) > max_width: padding = " ... " n = (max_width - len(padding)) // 2 q = (max_width - len(padding)) % 2 if q == 0: return "".join([s[0:n], padding, s[-n:]]) else: return "".join([s[0:n+1], padding, s[-n:]]) else: return s
[docs]def is_indexbox(box: list, errors="raise"): """ Check if this array matches a 2d or 3d index box definition Argopy expects one of the following 2 format to define an index box: - box = [lon_min, lon_max, lat_min, lat_max] - box = [lon_min, lon_max, lat_min, lat_max, datim_min, datim_max] This function check for this format compliance. Parameters ---------- box: list errors: str, default='raise' Returns ------- bool """ def is_dateconvertible(d): try: pd.to_datetime(d) isit = True except Exception: isit = False return isit tests = {} # Formats: tests["index box must be a list"] = lambda b: isinstance(b, list) tests["index box must be a list with 4 or 6 elements"] = lambda b: len(b) in [4, 6] # Types: tests["lon_min must be numeric"] = lambda b: ( isinstance(b[0], int) or isinstance(b[0], (np.floating, float)) ) tests["lon_max must be numeric"] = lambda b: ( isinstance(b[1], int) or isinstance(b[1], (np.floating, float)) ) tests["lat_min must be numeric"] = lambda b: ( isinstance(b[2], int) or isinstance(b[2], (np.floating, float)) ) tests["lat_max must be numeric"] = lambda b: ( isinstance(b[3], int) or isinstance(b[3], (np.floating, float)) ) if len(box) > 4: tests[ "datetim_min must be a string convertible to a Pandas datetime" ] = lambda b: isinstance(b[-2], str) and is_dateconvertible(b[-2]) tests[ "datetim_max must be a string convertible to a Pandas datetime" ] = lambda b: isinstance(b[-1], str) and is_dateconvertible(b[-1]) # Ranges: tests["lon_min must be in [-180;180] or [0;360]"] = ( lambda b: b[0] >= -180.0 and b[0] <= 360.0 ) tests["lon_max must be in [-180;180] or [0;360]"] = ( lambda b: b[1] >= -180.0 and b[1] <= 360.0 ) tests["lat_min must be in [-90;90]"] = lambda b: b[2] >= -90.0 and b[2] <= 90 tests["lat_max must be in [-90;90]"] = lambda b: b[3] >= -90.0 and b[3] <= 90.0 # Orders: tests["lon_max must be larger than lon_min"] = lambda b: b[0] < b[1] tests["lat_max must be larger than lat_min"] = lambda b: b[2] < b[3] if len(box) > 4: tests["datetim_max must come after datetim_min"] = lambda b: pd.to_datetime( b[-2] ) < pd.to_datetime(b[-1]) error = None for msg, test in tests.items(): if not test(box): error = msg break if error and errors == "raise": raise ValueError("%s: %s" % (box, error)) elif error: return False else: return True
[docs]def is_box(box: list, errors="raise"): """Check if this array matches a 3d or 4d data box definition Argopy expects one of the following 2 format to define a box: - box = [lon_min, lon_max, lat_min, lat_max, pres_min, pres_max] - box = [lon_min, lon_max, lat_min, lat_max, pres_min, pres_max, datim_min, datim_max] This function check for this format compliance. Parameters ---------- box: list errors: 'raise' Returns ------- bool """ def is_dateconvertible(d): try: pd.to_datetime(d) isit = True except Exception: isit = False return isit tests = {} # print(box) # Formats: tests["box must be a list"] = lambda b: isinstance(b, list) tests["box must be a list with 6 or 8 elements"] = lambda b: len(b) in [6, 8] # Types: tests["lon_min must be numeric"] = lambda b: ( isinstance(b[0], int) or isinstance(b[0], (np.floating, float)) ) tests["lon_max must be numeric"] = lambda b: ( isinstance(b[1], int) or isinstance(b[1], (np.floating, float)) ) tests["lat_min must be numeric"] = lambda b: ( isinstance(b[2], int) or isinstance(b[2], (np.floating, float)) ) tests["lat_max must be numeric"] = lambda b: ( isinstance(b[3], int) or isinstance(b[3], (np.floating, float)) ) tests["pres_min must be numeric"] = lambda b: ( isinstance(b[4], int) or isinstance(b[4], (np.floating, float)) ) tests["pres_max must be numeric"] = lambda b: ( isinstance(b[5], int) or isinstance(b[5], (np.floating, float)) ) if len(box) == 8: tests[ "datetim_min must be an object convertible to a Pandas datetime" ] = lambda b: is_dateconvertible(b[-2]) tests[ "datetim_max must be an object convertible to a Pandas datetime" ] = lambda b: is_dateconvertible(b[-1]) # Ranges: tests["lon_min must be in [-180;180] or [0;360]"] = ( lambda b: b[0] >= -180.0 and b[0] <= 360.0 ) tests["lon_max must be in [-180;180] or [0;360]"] = ( lambda b: b[1] >= -180.0 and b[1] <= 360.0 ) tests["lat_min must be in [-90;90]"] = lambda b: b[2] >= -90.0 and b[2] <= 90 tests["lat_max must be in [-90;90]"] = lambda b: b[3] >= -90.0 and b[3] <= 90.0 tests["pres_min must be in [0;10000]"] = lambda b: b[4] >= 0 and b[4] <= 10000 tests["pres_max must be in [0;10000]"] = lambda b: b[5] >= 0 and b[5] <= 10000 # Orders: tests["lon_max must be larger than lon_min"] = lambda b: b[0] <= b[1] tests["lat_max must be larger than lat_min"] = lambda b: b[2] <= b[3] tests["pres_max must be larger than pres_min"] = lambda b: b[4] <= b[5] if len(box) == 8: tests["datetim_max must come after datetim_min"] = lambda b: pd.to_datetime( b[-2] ) <= pd.to_datetime(b[-1]) error = None for msg, test in tests.items(): if not test(box): error = msg break if error and errors == "raise": raise ValueError("%s: %s" % (box, error)) elif error: return False else: return True
def is_list_of_strings(lst): return isinstance(lst, list) and all(isinstance(elem, str) for elem in lst) def is_list_of_dicts(lst): return all(isinstance(x, dict) for x in lst) def is_list_of_datasets(lst): return all(isinstance(x, xr.Dataset) for x in lst) def is_list_equal(lst1, lst2): """ Return true if 2 lists contain same elements""" return len(lst1) == len(lst2) and len(lst1) == sum( [1 for i, j in zip(lst1, lst2) if i == j] ) def to_list(obj): """Make sure that an expected list is indeed a list""" if not isinstance(obj, list): if isinstance(obj, np.ndarray): obj = list(obj) else: obj = [obj] return obj
[docs]def check_wmo(lst, errors="raise"): """ Validate a WMO option and returned it as a list of integers Parameters ---------- wmo: int WMO must be an integer or an iterable with elements that can be casted as integers errors: {'raise', 'warn', 'ignore'} Possibly raises a ValueError exception or UserWarning, otherwise fails silently. Returns ------- list(int) """ is_wmo(lst, errors=errors) # Make sure we deal with a list lst = to_list(lst) # Then cast list elements as integers return [abs(int(x)) for x in lst]
[docs]def is_wmo(lst, errors="raise"): # noqa: C901 """ Check if a WMO is valid Parameters ---------- wmo: int, list(int), array(int) WMO must be a single or a list of 5/7 digit positive numbers errors: {'raise', 'warn', 'ignore'} Possibly raises a ValueError exception or UserWarning, otherwise fails silently. Returns ------- bool True if wmo is indeed a list of integers """ # Make sure we deal with a list lst = to_list(lst) # Error message: # msg = "WMO must be an integer or an iterable with elements that can be casted as integers" msg = "WMO must be a single or a list of 5/7 digit positive numbers. Invalid: '{}'".format # Then try to cast list elements as integers, return True if ok result = True try: for x in lst: if not str(x).isdigit(): result = False if (len(str(x)) != 5) and (len(str(x)) != 7): result = False if int(x) <= 0: result = False except Exception: result = False if errors == "raise": raise ValueError(msg(x)) elif errors == 'warn': warnings.warn(msg(x)) if not result: if errors == "raise": raise ValueError(msg(x)) elif errors == 'warn': warnings.warn(msg(x)) else: return result
[docs]def check_cyc(lst, errors="raise"): """ Validate a CYC option and returned it as a list of integers Parameters ---------- cyc: int CYC must be an integer or an iterable with elements that can be casted as positive integers errors: {'raise', 'warn', 'ignore'} Possibly raises a ValueError exception or UserWarning, otherwise fails silently. Returns ------- list(int) """ is_cyc(lst, errors=errors) # Make sure we deal with a list lst = to_list(lst) # Then cast list elements as integers return [abs(int(x)) for x in lst]
[docs]def is_cyc(lst, errors="raise"): # noqa: C901 """ Check if a CYC is valid Parameters ---------- cyc: int, list(int), array(int) CYC must be a single or a list of at most 4 digit positive numbers errors: {'raise', 'warn', 'ignore'} Possibly raises a ValueError exception or UserWarning, otherwise fails silently. Returns ------- bool True if cyc is indeed a list of integers """ # Make sure we deal with a list lst = to_list(lst) # Error message: msg = "CYC must be a single or a list of at most 4 digit positive numbers. Invalid: '{}'".format # Then try to cast list elements as integers, return True if ok result = True try: for x in lst: if not str(x).isdigit(): result = False if (len(str(x)) > 4): result = False if int(x) < 0: result = False except Exception: result = False if errors == "raise": raise ValueError(msg(x)) elif errors == 'warn': warnings.warn(msg(x)) if not result: if errors == "raise": raise ValueError(msg(x)) elif errors == 'warn': warnings.warn(msg(x)) else: return result
def check_index_cols(column_names: list, convention: str = 'ar_index_global_prof'): """ ar_index_global_prof.txt: Index of profile files Profile directory file of the Argo Global Data Assembly Center file,date,latitude,longitude,ocean,profiler_type,institution,date_update argo_bio-profile_index.txt: bgc Argo profiles index file The directory file describes all individual bio-profile files of the argo GDAC ftp site. file,date,latitude,longitude,ocean,profiler_type,institution,parameters,parameter_data_mode,date_update """ # Default for 'ar_index_global_prof' ref = ['file', 'date', 'latitude', 'longitude', 'ocean', 'profiler_type', 'institution', 'date_update'] if convention == 'argo_bio-profile_index' or convention == 'argo_synthetic-profile_index': ref = ['file', 'date', 'latitude', 'longitude', 'ocean', 'profiler_type', 'institution', 'parameters', 'parameter_data_mode', 'date_update'] if not is_list_equal(column_names, ref): # log.debug("Expected: %s, got: %s" % (";".join(ref), ";".join(column_names))) raise InvalidDatasetStructure("Unexpected column names in this index !") else: return column_names def warnUnless(ok, txt): """Function to raise a warning unless condition is True This function IS NOT to be used as a decorator anymore Parameters ---------- ok: bool Condition to raise the warning or not txt: str Text to display in the warning """ if not ok: msg = "%s %s" % (inspect.stack()[1].function, txt) warnings.warn(msg) @contextlib.contextmanager def modified_environ(*remove, **update): """ Temporarily updates the ``os.environ`` dictionary in-place. The ``os.environ`` dictionary is updated in-place so that the modification is sure to work in all situations. :param remove: Environment variables to remove. :param update: Dictionary of environment variables and values to add/update. """ # Source: https://github.com/laurent-laporte-pro/stackoverflow-q2059482 env = os.environ update = update or {} remove = remove or [] # List of environment variables being updated or removed. stomped = (set(update.keys()) | set(remove)) & set(env.keys()) # Environment variables and values to restore on exit. update_after = {k: env[k] for k in stomped} # Environment variables and values to remove on exit. remove_after = frozenset(k for k in update if k not in env) try: env.update(update) [env.pop(k, None) for k in remove] yield finally: env.update(update_after) [env.pop(k) for k in remove_after] def toYearFraction( this_date: pd._libs.tslibs.timestamps.Timestamp = pd.to_datetime("now", utc=True) ): """ Compute decimal year, robust to leap years, precision to the second Compute the fraction of the year a given timestamp corresponds to. The "fraction of the year" goes: - from 0 on 01-01T00:00:00.000 of the year - to 1 on the 01-01T00:00:00.000 of the following year 1 second corresponds to the number of days in the year times 86400. The faction of the year is rounded to 10-digits in order to have a "second" precision. See discussion here: https://github.com/euroargodev/argodmqc_owc/issues/35 Parameters ---------- pd._libs.tslibs.timestamps.Timestamp Returns ------- float """ if "UTC" in [this_date.tzname() if this_date.tzinfo is not None else ""]: startOfThisYear = pd.to_datetime("%i-01-01T00:00:00.000" % this_date.year, utc=True) else: startOfThisYear = pd.to_datetime("%i-01-01T00:00:00.000" % this_date.year) yearDuration_sec = ( startOfThisYear + pd.offsets.DateOffset(years=1) - startOfThisYear ).total_seconds() yearElapsed_sec = (this_date - startOfThisYear).total_seconds() fraction = yearElapsed_sec / yearDuration_sec fraction = np.round(fraction, 10) return this_date.year + fraction def YearFraction_to_datetime(yf: float): """ Compute datetime from year fraction Inverse the toYearFraction() function Parameters ---------- float Returns ------- pd._libs.tslibs.timestamps.Timestamp """ year = np.int32(yf) fraction = yf - year fraction = np.round(fraction, 10) startOfThisYear = pd.to_datetime("%i-01-01T00:00:00" % year) yearDuration_sec = ( startOfThisYear + pd.offsets.DateOffset(years=1) - startOfThisYear ).total_seconds() yearElapsed_sec = pd.Timedelta(fraction * yearDuration_sec, unit="s") return pd.to_datetime(startOfThisYear + yearElapsed_sec, unit="s") def wrap_longitude(grid_long): """ Allows longitude (0-360) to wrap beyond the 360 mark, for mapping purposes. Makes sure that, if the longitude is near the boundary (0 or 360) that we wrap the values beyond 360 so it appears nicely on a map This is a refactor between get_region_data and get_region_hist_locations to avoid duplicate code source: https://github.com/euroargodev/argodmqc_owc/blob/e174f4538fdae1534c9740491398972b1ffec3ca/pyowc/utilities.py#L80 Parameters ---------- grid_long: array of longitude values Returns ------- array of longitude values that can extend past 360 """ neg_long = np.argwhere(grid_long < 0) grid_long[neg_long] = grid_long[neg_long] + 360 # if we have data close to upper boundary (360), then wrap some of the data round # so it appears on the map top_long = np.argwhere(grid_long >= 320) if top_long.__len__() != 0: bottom_long = np.argwhere(grid_long <= 40) grid_long[bottom_long] = 360 + grid_long[bottom_long] return grid_long
[docs]def wmo2box(wmo_id: int): """ Convert WMO square box number into a latitude/longitude box See: https://en.wikipedia.org/wiki/World_Meteorological_Organization_squares https://commons.wikimedia.org/wiki/File:WMO-squares-global.gif Parameters ---------- wmo_id: int WMO square number, must be between 1000 and 7817 Returns ------- box: list(int) [lon_min, lon_max, lat_min, lat_max] bounds to the WMO square number """ if wmo_id < 1000 or wmo_id > 7817: raise ValueError("Invalid WMO square number, must be between 1000 and 7817.") wmo_id = str(wmo_id) # "global quadrant" numbers where 1=NE, 3=SE, 5=SW, 7=NW quadrant = int(wmo_id[0]) if quadrant not in [1, 3, 5, 7]: raise ValueError("Invalid WMO square number, 1st digit must be 1, 3, 5 or 7.") # 'minimum' Latitude square boundary, nearest to the Equator nearest_to_the_Equator_latitude = int(wmo_id[1]) # 'minimum' Longitude square boundary, nearest to the Prime Meridian nearest_to_the_Prime_Meridian = int(wmo_id[2:4]) # dd = 10 if quadrant in [1, 3]: lon_min = nearest_to_the_Prime_Meridian * dd lon_max = nearest_to_the_Prime_Meridian * dd + dd elif quadrant in [5, 7]: lon_min = -nearest_to_the_Prime_Meridian * dd - dd lon_max = -nearest_to_the_Prime_Meridian * dd if quadrant in [1, 7]: lat_min = nearest_to_the_Equator_latitude * dd lat_max = nearest_to_the_Equator_latitude * dd + dd elif quadrant in [3, 5]: lat_min = -nearest_to_the_Equator_latitude * dd - dd lat_max = -nearest_to_the_Equator_latitude * dd box = [lon_min, lon_max, lat_min, lat_max] return box
[docs]def groupby_remap(z, data, z_regridded, # noqa C901 z_dim=None, z_regridded_dim="regridded", output_dim="remapped", select='deep', right=False): """ todo: Need a docstring here !""" # sub-sampling called in xarray ufunc def _subsample_bins(x, y, target_values): # remove all nans from input x and y idx = np.logical_or(np.isnan(x), np.isnan(y)) x = x[~idx] y = y[~idx] ifound = np.digitize( x, target_values, right=right ) # ``bins[i-1] <= x < bins[i]`` ifound -= 1 # Because digitize returns a 1-based indexing, we need to remove 1 y_binned = np.ones_like(target_values) * np.nan for ib, this_ibin in enumerate(np.unique(ifound)): ix = np.where(ifound == this_ibin) iselect = ix[-1] # Map to y value at specific x index in the bin: if select == "shallow": iselect = iselect[0] # min/shallow mapped_value = y[iselect] elif select == "deep": iselect = iselect[-1] # max/deep mapped_value = y[iselect] elif select == "middle": iselect = iselect[ np.where(x[iselect] >= np.median(x[iselect]))[0][0] ] # median/middle mapped_value = y[iselect] elif select == "random": iselect = iselect[np.random.randint(len(iselect))] mapped_value = y[iselect] # or Map to y statistics in the bin: elif select == "mean": mapped_value = np.nanmean(y[iselect]) elif select == "min": mapped_value = np.nanmin(y[iselect]) elif select == "max": mapped_value = np.nanmax(y[iselect]) elif select == "median": mapped_value = np.median(y[iselect]) else: raise InvalidOption("`select` option has invalid value (%s)" % select) y_binned[this_ibin] = mapped_value return y_binned # infer dim from input if z_dim is None: if len(z.dims) != 1: raise RuntimeError("if z_dim is not specified, x must be a 1D array.") dim = z.dims[0] else: dim = z_dim # if dataset is passed drop all data_vars that dont contain dim if isinstance(data, xr.Dataset): raise ValueError("Dataset input is not supported yet") # TODO: for a dataset input just apply the function for each appropriate array if version.parse(xr.__version__) > version.parse("0.15.0"): kwargs = dict( input_core_dims=[[dim], [dim], [z_regridded_dim]], output_core_dims=[[output_dim]], vectorize=True, dask="parallelized", output_dtypes=[data.dtype], dask_gufunc_kwargs={ "output_sizes": {output_dim: len(z_regridded[z_regridded_dim])} }, ) else: kwargs = dict( input_core_dims=[[dim], [dim], [z_regridded_dim]], output_core_dims=[[output_dim]], vectorize=True, dask="parallelized", output_dtypes=[data.dtype], output_sizes={output_dim: len(z_regridded[z_regridded_dim])}, ) remapped = xr.apply_ufunc(_subsample_bins, z, data, z_regridded, **kwargs) remapped.coords[output_dim] = z_regridded.rename( {z_regridded_dim: output_dim} ).coords[output_dim] return remapped
[docs]class TopoFetcher: """ Fetch topographic data through an ERDDAP server for an ocean rectangle Example: >>> from argopy import TopoFetcher >>> box = [-75, -45, 20, 30] # Lon_min, lon_max, lat_min, lat_max >>> ds = TopoFetcher(box).to_xarray() >>> ds = TopoFetcher(box, ds='gebco', stride=[10, 10], cache=True).to_xarray() """ class ERDDAP: def __init__(self, server: str, protocol: str = "tabledap"): self.server = server self.protocol = protocol self.response = "nc" self.dataset_id = "" self.constraints = ""
[docs] def __init__( self, box: list, ds: str = "gebco", cache: bool = False, cachedir: str = "", api_timeout: int = 0, stride: list = [1, 1], server: Union[str] = None, **kwargs, ): """ Instantiate an ERDDAP topo data fetcher Parameters ---------- ds: str (optional), default: 'gebco' Dataset to load: - 'gebco' will load the GEBCO_2020 Grid, a continuous terrain model for oceans and land at 15 arc-second intervals stride: list, default [1, 1] Strides along longitude and latitude. This allows to change the output resolution cache: bool (optional) Cache data or not (default: False) cachedir: str (optional) Path to cache folder api_timeout: int (optional) Erddap request time out in seconds. Set to OPTIONS['api_timeout'] by default. """ from .stores import httpstore timeout = OPTIONS["api_timeout"] if api_timeout == 0 else api_timeout self.fs = httpstore( cache=cache, cachedir=cachedir, timeout=timeout, size_policy="head" ) self.definition = "Erddap topographic data fetcher" self.BOX = box self.stride = stride if ds == "gebco": self.definition = "NOAA erddap gebco data fetcher for a space region" self.server = server if server is not None else "https://coastwatch.pfeg.noaa.gov/erddap" self.server_name = "NOAA" self.dataset_id = "gebco" self._init_erddap()
def _init_erddap(self): # Init erddap self.erddap = self.ERDDAP(server=self.server, protocol="griddap") self.erddap.response = "nc" if self.dataset_id == "gebco": self.erddap.dataset_id = "GEBCO_2020" else: raise ValueError( "Invalid database short name for %s erddap" % self.server_name ) return self def _cname(self) -> str: """ Fetcher one line string definition helper """ cname = "?" if hasattr(self, "BOX"): BOX = self.BOX cname = ("[x=%0.2f/%0.2f; y=%0.2f/%0.2f]") % ( BOX[0], BOX[1], BOX[2], BOX[3], ) return cname def __repr__(self): summary = ["<topofetcher.erddap>"] summary.append("Name: %s" % self.definition) summary.append("API: %s" % self.server) summary.append("Domain: %s" % format_oneline(self.cname())) return "\n".join(summary)
[docs] def cname(self): """ Return a unique string defining the constraints """ return self._cname()
@property def cachepath(self): """ Return path to cached file(s) for this request Returns ------- list(str) """ return [self.fs.cachepath(uri) for uri in self.uri]
[docs] def define_constraints(self): """ Define request constraints """ # Eg: https://coastwatch.pfeg.noaa.gov/erddap/griddap/GEBCO_2020.nc?elevation%5B(34):5:(42)%5D%5B(-21):7:(-12)%5D self.erddap.constraints = "%s(%0.2f):%i:(%0.2f)%s%s(%0.2f):%i:(%0.2f)%s" % ( "%5B", self.BOX[2], self.stride[1], self.BOX[3], "%5D", "%5B", self.BOX[0], self.stride[0], self.BOX[1], "%5D", ) return None
# @property # def _minimal_vlist(self): # """ Return the minimal list of variables to retrieve """ # vlist = list() # vlist.append("latitude") # vlist.append("longitude") # vlist.append("elevation") # return vlist def url_encode(self, url): """ Return safely encoded list of urls This is necessary because fsspec cannot handle in cache paths/urls with a '[' character """ # return urls def safe_for_fsspec_cache(url): url = url.replace("[", "%5B") # This is the one really necessary url = url.replace("]", "%5D") # For consistency return url return safe_for_fsspec_cache(url)
[docs] def get_url(self): """ Return the URL to download data requested Returns ------- str """ # First part of the URL: protocol = self.erddap.protocol dataset_id = self.erddap.dataset_id response = self.erddap.response url = f"{self.erddap.server}/{protocol}/{dataset_id}.{response}?" # Add variables to retrieve: variables = ["elevation"] variables = ",".join(variables) url += f"{variables}" # Add constraints: self.define_constraints() # Define constraint to select this box of data (affect self.erddap.constraints) url += f"{self.erddap.constraints}" return self.url_encode(url)
@property def uri(self): """ List of files to load for a request Returns ------- list(str) """ return [self.get_url()]
[docs] def to_xarray(self, errors: str = "ignore"): """ Load Topographic data and return a xarray.DataSet """ # Download data if len(self.uri) == 1: ds = self.fs.open_dataset(self.uri[0]) return ds
[docs] def load(self, errors: str = "ignore"): """ Load Topographic data and return a xarray.DataSet """ return self.to_xarray(errors=errors)
def argo_split_path(this_path): # noqa C901 """ Split path from a GDAC ftp style Argo netcdf file and return information >>> argo_split_path('coriolis/6901035/profiles/D6901035_001D.nc') >>> argo_split_path('https://data-argo.ifremer.fr/dac/csiro/5903939/profiles/D5903939_103.nc') Parameters ---------- str Returns ------- dict """ dacs = [ "aoml", "bodc", "coriolis", "csio", "csiro", "incois", "jma", "kma", "kordi", "meds", "nmdis", ] output = {} start_with = lambda f, x: f[0:len(x)] == x if len(x) <= len(f) else False # noqa: E731 def split_path(p, sep='/'): """Split a pathname. Returns tuple "(head, tail)" where "tail" is everything after the final slash. Either part may be empty.""" # Same as posixpath.py but we get to choose the file separator ! p = os.fspath(p) i = p.rfind(sep) + 1 head, tail = p[:i], p[i:] if head and head != sep * len(head): head = head.rstrip(sep) return head, tail def fix_localhost(host): if 'ftp://localhost:' in host: return "ftp://%s" % (urlparse(host).netloc) if 'http://127.0.0.1:' in host: return "http://%s" % (urlparse(host).netloc) else: return "" known_origins = ['https://data-argo.ifremer.fr', 'ftp://ftp.ifremer.fr/ifremer/argo', 'ftp://usgodae.org/pub/outgoing/argo', fix_localhost(this_path), ''] output['origin'] = [origin for origin in known_origins if start_with(this_path, origin)][0] output['origin'] = '.' if output['origin'] == '' else output['origin'] + '/' sep = '/' if output['origin'] != '.' else os.path.sep (path, file) = split_path(this_path, sep=sep) output['path'] = path.replace(output['origin'], '') output['name'] = file # Deal with the path: # dac/<DAC>/<FloatWmoID>/ # dac/<DAC>/<FloatWmoID>/profiles path_parts = path.split(sep) try: if path_parts[-1] == 'profiles': output['type'] = 'Mono-cycle profile file' output['wmo'] = path_parts[-2] output['dac'] = path_parts[-3] else: output['type'] = 'Multi-cycle profile file' output['wmo'] = path_parts[-1] output['dac'] = path_parts[-2] except Exception: log.warning(this_path) log.warning(path) log.warning(sep) log.warning(path_parts) log.warning(output) raise if output['dac'] not in dacs: log.debug("This is not a Argo GDAC compliant file path: %s" % path) log.warning(this_path) log.warning(path) log.warning(sep) log.warning(path_parts) log.warning(output) raise ValueError("This is not a Argo GDAC compliant file path (invalid DAC name: '%s')" % output['dac']) # Deal with the file name: filename, file_extension = os.path.splitext(output['name']) output['extension'] = file_extension if file_extension != '.nc': raise ValueError( "This is not a Argo GDAC compliant file path (invalid file extension: '%s')" % file_extension) filename_parts = output['name'].split("_") if "Mono" in output['type']: prefix = filename_parts[0].split(output['wmo'])[0] if 'R' in prefix: output['data_mode'] = 'R, Real-time data' if 'D' in prefix: output['data_mode'] = 'D, Delayed-time data' if 'S' in prefix: output['type'] = 'S, Synthetic BGC Mono-cycle profile file' if 'M' in prefix: output['type'] = 'M, Merged BGC Mono-cycle profile file' if 'B' in prefix: output['type'] = 'B, BGC Mono-cycle profile file' suffix = filename_parts[-1].split(output['wmo'])[-1] if 'D' in suffix: output['direction'] = 'D, descending profiles' elif suffix == "" and "Mono" in output['type']: output['direction'] = 'A, ascending profiles (implicit)' else: typ = filename_parts[-1].split(".nc")[0] if typ == 'prof': output['type'] = 'Multi-cycle file' if typ == 'Sprof': output['type'] = 'S, Synthetic BGC Multi-cycle profiles file' if typ == 'tech': output['type'] = 'Technical data file' if typ == 'meta': output['type'] = 'Metadata file' if 'traj' in typ: output['type'] = 'Trajectory file' if typ.split("traj")[0] == 'D': output['data_mode'] = 'D, Delayed-time data' elif typ.split("traj")[0] == 'R': output['data_mode'] = 'R, Real-time data' else: output['data_mode'] = 'R, Real-time data (implicit)' # Adjust origin and path for local files: # This ensure that output['path'] is agnostic to users and can be reused on any gdac compliant architecture parts = path.split(sep) i, stop = len(parts) - 1, False while not stop: if parts[i] == 'profiles' or parts[i] == output['wmo'] or parts[i] == output['dac'] or parts[i] == 'dac': i = i - 1 if i < 0: stop = True else: stop = True output['origin'] = sep.join(parts[0:i + 1]) output['path'] = output['path'].replace(output['origin'], '') return dict(sorted(output.items())) class DocInherit(object): """Docstring inheriting method descriptor The class itself is also used as a decorator Usage: class Foo(object): def foo(self): "Frobber" pass class Bar(Foo): @doc_inherit def foo(self): pass Now, Bar.foo.__doc__ == Bar().foo.__doc__ == Foo.foo.__doc__ == "Frobber" src: https://code.activestate.com/recipes/576862/ """ def __init__(self, mthd): self.mthd = mthd self.name = mthd.__name__ def __get__(self, obj, cls): if obj: return self.get_with_inst(obj, cls) else: return self.get_no_inst(cls) def get_with_inst(self, obj, cls): overridden = getattr(super(cls, obj), self.name, None) @wraps(self.mthd, assigned=('__name__', '__module__')) def f(*args, **kwargs): return self.mthd(obj, *args, **kwargs) return self.use_parent_doc(f, overridden) def get_no_inst(self, cls): for parent in cls.__mro__[1:]: overridden = getattr(parent, self.name, None) if overridden: break @wraps(self.mthd, assigned=('__name__', '__module__')) def f(*args, **kwargs): return self.mthd(*args, **kwargs) return self.use_parent_doc(f, overridden) def use_parent_doc(self, func, source): if source is None: raise NameError("Can't find '%s' in parents" % self.name) func.__doc__ = source.__doc__ return func doc_inherit = DocInherit
[docs]def deprecated(reason): """Deprecation warning decorator. This is a decorator which can be used to mark functions as deprecated. It will result in a warning being emitted when the function is used. Parameters ---------- reason: {str, None} Text message to send with deprecation warning Examples -------- The @deprecated can be used with a 'reason'. .. code-block:: python @deprecated("please, use another function") def old_function(x, y): pass or without: .. code-block:: python @deprecated def old_function(x, y): pass References ---------- https://stackoverflow.com/a/40301488 """ import inspect if isinstance(reason, str): def decorator(func1): if inspect.isclass(func1): fmt1 = "Call to deprecated class {name} ({reason})." else: fmt1 = "Call to deprecated function {name} ({reason})." @wraps(func1) def new_func1(*args, **kwargs): warnings.simplefilter('always', DeprecationWarning) warnings.warn( fmt1.format(name=func1.__name__, reason=reason), category=DeprecationWarning, stacklevel=2 ) warnings.simplefilter('default', DeprecationWarning) return func1(*args, **kwargs) return new_func1 return decorator elif inspect.isclass(reason) or inspect.isfunction(reason): func2 = reason if inspect.isclass(func2): fmt2 = "Call to deprecated class {name}." else: fmt2 = "Call to deprecated function {name}." @wraps(func2) def new_func2(*args, **kwargs): warnings.simplefilter('always', DeprecationWarning) warnings.warn( fmt2.format(name=func2.__name__), category=DeprecationWarning, stacklevel=2 ) warnings.simplefilter('default', DeprecationWarning) return func2(*args, **kwargs) return new_func2 else: raise TypeError(repr(type(reason)))
class RegistryItem(ABC): """Prototype for possible custom items in a Registry""" @property @abstractmethod def value(self): raise NotImplementedError("Not implemented") @property @abstractmethod def isvalid(self, item): raise NotImplementedError("Not implemented") @abstractmethod def __str__(self): raise NotImplementedError("Not implemented") @abstractmethod def __repr__(self): raise NotImplementedError("Not implemented")
[docs]class float_wmo(RegistryItem): """Argo float WMO number object"""
[docs] def __init__(self, WMO_number, errors='raise'): """Create an Argo float WMO number object Parameters ---------- WMO_number: object Anything that could be casted as an integer errors: {'raise', 'warn', 'ignore'} Possibly raises a ValueError exception or UserWarning, otherwise fails silently if WMO_number is not valid Returns ------- :class:`argopy.utilities.float_wmo` """ self.errors = errors if isinstance(WMO_number, float_wmo): item = WMO_number.value else: item = check_wmo(WMO_number, errors=self.errors)[0] # This will automatically validate item self.item = item
@property def isvalid(self): """Check if WMO number is valid""" return is_wmo(self.item, errors=self.errors) # return True # Because it was checked at instantiation @property def value(self): """Return WMO number as in integer""" return int(self.item) def __str__(self): # return "%s" % check_wmo(self.item)[0] return "%s" % self.item def __repr__(self): return f"WMO({self.item})" def __check_other__(self, other): return check_wmo(other)[0] if type(other) is not float_wmo else other.item def __eq__(self, other): return self.item.__eq__(self.__check_other__(other)) def __ne__(self, other): return self.item.__ne__(self.__check_other__(other)) def __gt__(self, other): return self.item.__gt__(self.__check_other__(other)) def __lt__(self, other): return self.item.__lt__(self.__check_other__(other)) def __ge__(self, other): return self.item.__ge__(self.__check_other__(other)) def __le__(self, other): return self.item.__le__(self.__check_other__(other)) def __hash__(self): return hash(self.item)
[docs]class Registry(UserList): """A list manager can that validate item type Examples -------- You can commit new entry to the registry, one by one: >>> R = Registry(name='file') >>> R.commit('meds/4901105/profiles/D4901105_017.nc') >>> R.commit('aoml/1900046/profiles/D1900046_179.nc') Or with a list: >>> R = Registry(name='My floats', dtype='wmo') >>> R.commit([2901746, 4902252]) And also at instantiation time (name and dtype are optional): >>> R = Registry([2901746, 4902252], name='My floats', dtype=float_wmo) Registry can be used like a list. It is iterable: >>> for wmo in R: >>> print(wmo) It has a ``len`` property: >>> len(R) It can be checked for values: >>> 4902252 in R You can also remove items from the registry, again one by one or with a list: >>> R.remove('2901746') """ def _complain(self, msg): if self._invalid == 'raise': raise ValueError(msg) elif self._invalid == 'warn': warnings.warn(msg) else: log.debug(msg) def _str(self, item): is_valid = isinstance(item, str) if not is_valid: self._complain("%s is not a valid %s" % (str(item), self.dtype)) return is_valid def _dict(self, item): is_valid = isinstance(item, dict) if not is_valid: self._complain("%s is not a valid %s" % (str(item), self.dtype)) return is_valid def _wmo(self, item): return item.isvalid
[docs] def __init__(self, initlist=None, name: str = 'unnamed', dtype='str', invalid='raise'): """Create a registry, i.e. a controlled list Parameters ---------- initlist: list, optional List of values to register name: str, default: 'unnamed' Name of the Registry dtype: :class:`str` or dtype, default: :class:`str` Data type of registry content. Supported values are: 'str', 'wmo', float_wmo invalid: str, default: 'raise' Define what do to when a new item is not valid. Can be 'raise' or 'ignore' """ self.name = name self._invalid = invalid if repr(dtype) == "<class 'str'>" or dtype == 'str': self._validator = self._str self.dtype = str elif dtype == float_wmo or str(dtype).lower() == 'wmo': self._validator = self._wmo self.dtype = float_wmo elif repr(dtype) == "<class 'dict'>" or dtype == 'dict': self._validator = self._dict self.dtype = dict else: raise ValueError("Unrecognised Registry data type '%s'" % dtype) if initlist is not None: initlist = self._process_items(initlist) super().__init__(initlist)
def __repr__(self): summary = ["<argopy.registry>%s" % str(self.dtype)] summary.append("Name: %s" % self.name) N = len(self.data) msg = "Nitems: %s" % N if N > 1 else "Nitem: %s" % N summary.append(msg) if N > 0: items = [repr(item) for item in self.data] # msg = format_oneline("[%s]" % "; ".join(items), max_width=120) msg = "[%s]" % "; ".join(items) summary.append("Content: %s" % msg) return "\n".join(summary) def _process_items(self, items): if not isinstance(items, list): items = [items] if self.dtype == float_wmo: items = [float_wmo(item, errors=self._invalid) for item in items] return items def commit(self, values): """R.commit(values) -- append values to the end of the registry if not already in""" items = self._process_items(values) for item in items: if item not in self.data and self._validator(item): super().append(item) return self def append(self, value): """R.append(value) -- append value to the end of the registry""" items = self._process_items(value) for item in items: if self._validator(item): super().append(item) return self def extend(self, other): """R.extend(iterable) -- extend registry by appending elements from the iterable""" self.append(other) return self def remove(self, values): """R.remove(valueS) -- remove first occurrence of values.""" items = self._process_items(values) for item in items: if item in self.data: super().remove(item) return self def insert(self, index, value): """R.insert(index, value) -- insert value before index.""" item = self._process_items(value)[0] if self._validator(item): super().insert(index, item) return self def __copy__(self): # Called with copy.copy(R) return Registry(copy.copy(self.data), dtype=self.dtype) def copy(self): """Return a shallow copy of the registry""" return self.__copy__()
[docs]def get_coriolis_profile_id(WMO, CYC=None, **kwargs): """ Return a :class:`pandas.DataFrame` with CORIOLIS ID of WMO/CYC profile pairs This method get ID by requesting the dataselection.euro-argo.eu trajectory API. Parameters ---------- WMO: int, list(int) Define the list of Argo floats. This is a list of integers with WMO float identifiers. WMO is the World Meteorological Organization. CYC: int, list(int) Define the list of cycle numbers to load ID for each Argo floats listed in ``WMO``. Returns ------- :class:`pandas.DataFrame` """ WMO_list = check_wmo(WMO) if CYC is not None: CYC_list = check_cyc(CYC) if 'api_server' in kwargs: api_server = kwargs['api_server'] elif OPTIONS['server'] is not None: api_server = OPTIONS['server'] else: api_server = "https://dataselection.euro-argo.eu/api" URIs = [api_server + "/trajectory/%i" % wmo for wmo in WMO_list] def prec(data, url): # Transform trajectory json to dataframe # See: https://dataselection.euro-argo.eu/swagger-ui.html#!/cycle-controller/getCyclesByPlatformCodeUsingGET WMO = check_wmo(url.split("/")[-1])[0] rows = [] for profile in data: keys = [x for x in profile.keys() if x not in ["coordinate"]] meta_row = dict((key, profile[key]) for key in keys) for row in profile["coordinate"]: meta_row[row] = profile["coordinate"][row] meta_row["WMO"] = WMO rows.append(meta_row) return pd.DataFrame(rows) from .stores import httpstore fs = httpstore(cache=True, cachedir=OPTIONS['cachedir']) data = fs.open_mfjson(URIs, preprocess=prec, errors="raise", url_follow=True) # Merge results (list of dataframe): key_map = { "id": "ID", "lat": "LATITUDE", "lon": "LONGITUDE", "cvNumber": "CYCLE_NUMBER", "level": "level", "WMO": "PLATFORM_NUMBER", } for i, df in enumerate(data): df = df.reset_index() df = df.rename(columns=key_map) df = df[[value for value in key_map.values() if value in df.columns]] data[i] = df df = pd.concat(data, ignore_index=True) df.sort_values(by=["PLATFORM_NUMBER", "CYCLE_NUMBER"], inplace=True) df = df.reset_index(drop=True) # df = df.set_index(["PLATFORM_NUMBER", "CYCLE_NUMBER"]) df = df.astype({"ID": int}) if CYC is not None: df = pd.concat([df[df["CYCLE_NUMBER"] == cyc] for cyc in CYC_list]).reset_index( drop=True ) return df[ ["PLATFORM_NUMBER", "CYCLE_NUMBER", "ID", "LATITUDE", "LONGITUDE", "level"] ]
[docs]def get_ea_profile_page(WMO, CYC=None, **kwargs): """ Return a list of URL Parameters ---------- WMO: int, list(int) WMO must be an integer or an iterable with elements that can be casted as integers CYC: int, list(int), default (None) CYC must be an integer or an iterable with elements that can be casted as positive integers Returns ------- list(str) See also -------- get_coriolis_profile_id """ df = get_coriolis_profile_id(WMO, CYC, **kwargs) url = "https://dataselection.euro-argo.eu/cycle/{}" return [url.format(this_id) for this_id in sorted(df["ID"])]
[docs]class ArgoNVSReferenceTables: """Argo Reference Tables Utility function to retrieve Argo Reference Tables from a NVS server. By default, this relies on: https://vocab.nerc.ac.uk/collection Examples -------- >>> R = ArgoNVSReferenceTables() >>> R.valid_ref >>> R.all_tbl_name() >>> R.tbl(3) >>> R.tbl('R09') >>> R.all_tbl() """ valid_ref = [ "R01", "RR2", "RD2", "RP2", "R03", "R04", "R05", "R06", "R07", "R08", "R09", "R10", "R11", "R12", "R13", "R15", "RMC", "RTV", "R16", # "R18", "R19", "R20", "R21", "R22", "R23", "R24", "R25", "R26", "R27", # "R28", # "R29", # "R30", ] """List of all available Reference Tables"""
[docs] def __init__(self, nvs="https://vocab.nerc.ac.uk/collection", cache: bool = True, cachedir: str = "", ): """Argo Reference Tables from NVS""" from .stores import httpstore cachedir = OPTIONS["cachedir"] if cachedir == "" else cachedir self.fs = httpstore(cache=cache, cachedir=cachedir) self.nvs = nvs
def _valid_ref(self, rtid): if rtid not in self.valid_ref: rtid = "R%0.2d" % rtid if rtid not in self.valid_ref: raise ValueError( "Invalid Argo Reference Table, should be one in: %s" % ", ".join(self.valid_ref) ) return rtid def _jsConcept2df(self, data): """Return all skos:Concept as class:`pandas.DataFrame`""" content = { "altLabel": [], "prefLabel": [], "definition": [], "deprecated": [], "id": [], } for k in data["@graph"]: if k["@type"] == "skos:Collection": Collection_name = k["alternative"] elif k["@type"] == "skos:Concept": content["altLabel"].append(k["altLabel"]) content["prefLabel"].append(k["prefLabel"]["@value"]) content["definition"].append(k["definition"]["@value"]) content["deprecated"].append(k["deprecated"]) content["id"].append(k["@id"]) df = pd.DataFrame.from_dict(content) df.name = Collection_name return df def _jsCollection(self, data): """Return last skos:Collection information as data""" for k in data["@graph"]: if k["@type"] == "skos:Collection": name = k["alternative"] desc = k["description"] rtid = k["@id"] return (name, desc, rtid) def get_url(self, rtid, fmt="ld+json"): """Return URL toward a given reference table for a given format Parameters ---------- rtid: {str, int} Name or number of the reference table to retrieve. Eg: 'R01', 12 fmt: str, default: "ld+json" Format of the NVS server response. Can be: "ld+json", "rdf+xml" or "text/turtle". Returns ------- str """ rtid = self._valid_ref(rtid) if fmt == "ld+json": fmt_ext = "?_profile=nvs&_mediatype=application/ld+json" elif fmt == "rdf+xml": fmt_ext = "?_profile=nvs&_mediatype=application/rdf+xml" elif fmt == "text/turtle": fmt_ext = "?_profile=nvs&_mediatype=text/turtle" else: raise ValueError("Invalid format. Must be in: 'ld+json', 'rdf+xml' or 'text/turtle'.") url = "{}/{}/current/{}".format return url(self.nvs, rtid, fmt_ext)
[docs] def tbl(self, rtid): """Return an Argo Reference table Parameters ---------- rtid: {str, int} Name or number of the reference table to retrieve. Eg: 'R01', 12 Returns ------- class:`pandas.DataFrame` """ rtid = self._valid_ref(rtid) js = self.fs.open_json(self.get_url(rtid)) df = self._jsConcept2df(js) return df
[docs] def tbl_name(self, rtid): """Return name of an Argo Reference table Parameters ---------- rtid: {str, int} Name or number of the reference table to retrieve. Eg: 'R01', 12 Returns ------- tuple('short name', 'description', 'NVS id link') """ rtid = self._valid_ref(rtid) js = self.fs.open_json(self.get_url(rtid)) return self._jsCollection(js)
@property def all_tbl(self): """Return all Argo Reference tables Returns ------- OrderedDict Dictionary with all table short names as key and table content as class:`pandas.DataFrame` """ URLs = [self.get_url(rtid) for rtid in self.valid_ref] df_list = self.fs.open_mfjson(URLs, preprocess=self._jsConcept2df) all_tables = {} [all_tables.update({t.name: t}) for t in df_list] all_tables = collections.OrderedDict(sorted(all_tables.items())) return all_tables @property def all_tbl_name(self): """Return names of all Argo Reference tables Returns ------- OrderedDict Dictionary with all table short names as key and table names as tuple('short name', 'description', 'NVS id link') """ URLs = [self.get_url(rtid) for rtid in self.valid_ref] name_list = self.fs.open_mfjson(URLs, preprocess=self._jsCollection) all_tables = {} [ all_tables.update({rtid.split("/")[-3]: (name, desc, rtid)}) for name, desc, rtid in name_list ] all_tables = collections.OrderedDict(sorted(all_tables.items())) return all_tables
[docs]class OceanOPSDeployments: """Use the OceanOPS API for metadata access to retrieve Argo floats deployment information. The API is documented here: https://www.ocean-ops.org/api/swagger/?url=https://www.ocean-ops.org/api/1/oceanops-api.yaml Description of deployment status name: =========== == ==== Status Id Description =========== == ==== PROBABLE 0 Starting status for some platforms, when there is only a few metadata available, like rough deployment location and date. The platform may be deployed CONFIRMED 1 Automatically set when a ship is attached to the deployment information. The platform is ready to be deployed, deployment is planned REGISTERED 2 Starting status for most of the networks, when deployment planning is not done. The deployment is certain, and a notification has been sent via the OceanOPS system OPERATIONAL 6 Automatically set when the platform is emitting a pulse and observations are distributed within a certain time interval INACTIVE 4 The platform is not emitting a pulse since a certain time CLOSED 5 The platform is not emitting a pulse since a long time, it is considered as dead =========== == ==== Examples -------- Import the utility class: >>> from argopy.utilities import OceanOPSDeployments >>> from argopy import OceanOPSDeployments Possibly define the space/time box to work with: >>> box = [-20, 0, 42, 51] >>> box = [-20, 0, 42, 51, '2020-01', '2021-01'] >>> box = [-180, 180, -90, 90, '2020-01', None] Instantiate the metadata fetcher: >>> deployment = OceanOPSDeployments() >>> deployment = OceanOPSDeployments(box) >>> deployment = OceanOPSDeployments(box, deployed_only=True) # Remove planification Load information: >>> df = deployment.to_dataframe() >>> data = deployment.to_json() Useful attributes and methods: >>> deployment.uri >>> deployment.uri_decoded >>> deployment.status_code >>> fig, ax = deployment.plot_status() >>> plan_virtualfleet = deployment.plan """ api = "https://www.ocean-ops.org" """URL to the API""" model = "api/1/data/platform" """This model represents a Platform entity and is used to retrieve a platform information (schema model named 'Ptf').""" api_server_check = 'https://www.ocean-ops.org/api/1/oceanops-api.yaml' """URL to check if the API is alive"""
[docs] def __init__(self, box: list = None, deployed_only: bool = False): """ Parameters ---------- box: list, optional, default=None Define the domain to load the Argo deployment plan for. By default, **box** is set to None to work with the global deployment plan starting from the current date. The list expects one of the following format: - [lon_min, lon_max, lat_min, lat_max] - [lon_min, lon_max, lat_min, lat_max, date_min] - [lon_min, lon_max, lat_min, lat_max, date_min, date_max] Longitude and latitude values must be floats. Dates are strings. If **box** is provided with a regional domain definition (only 4 values given), then ``date_min`` will be set to the current date. deployed_only: bool, optional, default=False Return only floats already deployed. If set to False (default), will return the full deployment plan (floats with all possible status). If set to True, will return only floats with one of the following status: ``OPERATIONAL``, ``INACTIVE``, and ``CLOSED``. """ if box is None: box = [None, None, None, None, pd.to_datetime('now', utc=True).strftime("%Y-%m-%d"), None] elif len(box) == 4: box.append(pd.to_datetime('now', utc=True).strftime("%Y-%m-%d")) box.append(None) elif len(box) == 5: box.append(None) if len(box) != 6: raise ValueError("The 'box' argument must be: None or of lengths 4 or 5 or 6\n%s" % str(box)) self.box = box self.deployed_only = deployed_only self.data = None from .stores import httpstore self.fs = httpstore(cache=False)
def __format(self, x, typ: str) -> str: """ string formatting helper """ if typ == "lon": return str(x) if x is not None else "-" elif typ == "lat": return str(x) if x is not None else "-" elif typ == "tim": return pd.to_datetime(x).strftime("%Y-%m-%d") if x is not None else "-" else: return str(x) def __repr__(self): summary = ["<argo.deployment_plan>"] summary.append("API: %s/%s" % (self.api, self.model)) summary.append("Domain: %s" % self.box_name) summary.append("Deployed only: %s" % self.deployed_only) if self.data is not None: summary.append("Nb of floats in the deployment plan: %s" % self.size) else: summary.append("Nb of floats in the deployment plan: - [Data not retrieved yet]") return '\n'.join(summary) def __encode_inc(self, inc): """Return encoded uri expression for 'include' parameter Parameters ---------- inc: str Returns ------- str """ return inc.replace("\"", "%22").replace("[", "%5B").replace("]", "%5D") def __encode_exp(self, exp): """Return encoded uri expression for 'exp' parameter Parameters ---------- exp: str Returns ------- str """ return exp.replace("\"", "%22").replace("'", "%27").replace(" ", "%20").replace(">", "%3E").replace("<", "%3C") def __get_uri(self, encoded=False): uri = "exp=%s&include=%s" % (self.exp(encoded=encoded), self.include(encoded=encoded)) url = "%s/%s?%s" % (self.api, self.model, uri) return url def include(self, encoded=False): """Return an Ocean-Ops API 'include' expression This is used to determine which variables the API call should return Parameters ---------- encoded: bool, default=False Returns ------- str """ # inc = ["ref", "ptfDepl.lat", "ptfDepl.lon", "ptfDepl.deplDate", "ptfStatus", "wmos"] # inc = ["ref", "ptfDepl.lat", "ptfDepl.lon", "ptfDepl.deplDate", "ptfStatus.id", "ptfStatus.name", "wmos"] # inc = ["ref", "ptfDepl.lat", "ptfDepl.lon", "ptfDepl.deplDate", "ptfStatus.id", "ptfStatus.name"] inc = ["ref", "ptfDepl.lat", "ptfDepl.lon", "ptfDepl.deplDate", "ptfStatus.id", "ptfStatus.name", "ptfStatus.description", "program.nameShort", "program.country.nameShort", "ptfModel.nameShort", "ptfDepl.noSite"] inc = "[%s]" % ",".join(["\"%s\"" % v for v in inc]) return inc if not encoded else self.__encode_inc(inc) def exp(self, encoded=False): """Return an Ocean-Ops API deployment search expression for an argopy region box definition Parameters ---------- encoded: bool, default=False Returns ------- str """ exp, arg = "networkPtfs.network.name='Argo'", [] if self.box[0] is not None: exp += " and ptfDepl.lon>=$var%i" % (len(arg) + 1) arg.append(str(self.box[0])) if self.box[1] is not None: exp += " and ptfDepl.lon<=$var%i" % (len(arg) + 1) arg.append(str(self.box[1])) if self.box[2] is not None: exp += " and ptfDepl.lat>=$var%i" % (len(arg) + 1) arg.append(str(self.box[2])) if self.box[3] is not None: exp += " and ptfDepl.lat<=$var%i" % (len(arg) + 1) arg.append(str(self.box[3])) if len(self.box) > 4: if self.box[4] is not None: exp += " and ptfDepl.deplDate>=$var%i" % (len(arg) + 1) arg.append("\"%s\"" % pd.to_datetime(self.box[4]).strftime("%Y-%m-%d %H:%M:%S")) if self.box[5] is not None: exp += " and ptfDepl.deplDate<=$var%i" % (len(arg) + 1) arg.append("\"%s\"" % pd.to_datetime(self.box[5]).strftime("%Y-%m-%d %H:%M:%S")) if self.deployed_only: exp += " and ptfStatus>=$var%i" % (len(arg) + 1) arg.append(str(4)) # Allow for: 4, 5 or 6 exp = "[\"%s\", %s]" % (exp, ", ".join(arg)) return exp if not encoded else self.__encode_exp(exp) @property def size(self): return len(self.data['data']) if self.data is not None else None @property def status_code(self): """Return a :class:`pandas.DataFrame` with the definition of status""" status = {'status_code': [0, 1, 2, 6, 4, 5], 'status_name': ['PROBABLE', 'CONFIRMED', 'REGISTERED', 'OPERATIONAL', 'INACTIVE', 'CLOSED'], 'description': [ 'Starting status for some platforms, when there is only a few metadata available, like rough deployment location and date. The platform may be deployed', 'Automatically set when a ship is attached to the deployment information. The platform is ready to be deployed, deployment is planned', 'Starting status for most of the networks, when deployment planning is not done. The deployment is certain, and a notification has been sent via the OceanOPS system', 'Automatically set when the platform is emitting a pulse and observations are distributed within a certain time interval', 'The platform is not emitting a pulse since a certain time', 'The platform is not emitting a pulse since a long time, it is considered as dead', ]} return pd.DataFrame(status).set_index('status_code') @property def box_name(self): """Return a string to print the box property""" BOX = self.box cname = ("[lon=%s/%s; lat=%s/%s]") % ( self.__format(BOX[0], "lon"), self.__format(BOX[1], "lon"), self.__format(BOX[2], "lat"), self.__format(BOX[3], "lat"), ) if len(BOX) == 6: cname = ("[lon=%s/%s; lat=%s/%s; t=%s/%s]") % ( self.__format(BOX[0], "lon"), self.__format(BOX[1], "lon"), self.__format(BOX[2], "lat"), self.__format(BOX[3], "lat"), self.__format(BOX[4], "tim"), self.__format(BOX[5], "tim"), ) return cname @property def uri(self): """Return encoded URL to post an Ocean-Ops API request Returns ------- str """ return self.__get_uri(encoded=True) @property def uri_decoded(self): """Return decoded URL to post an Ocean-Ops API request Returns ------- str """ return self.__get_uri(encoded=False) @property def plan(self): """Return a dictionary to be used as argument in a :class:`virtualargofleet.VirtualFleet` This method is for dev, but will be moved to the VirtualFleet software utilities """ df = self.to_dataframe() plan = df[['lon', 'lat', 'date']].rename(columns={"date": "time"}).to_dict('series') for key in plan.keys(): plan[key] = plan[key].to_list() plan['time'] = np.array(plan['time'], dtype='datetime64') return plan def to_json(self): """Return OceanOPS API request response as a json object""" if self.data is None: self.data = self.fs.open_json(self.uri) return self.data
[docs] def to_dataframe(self): """Return the deployment plan as :class:`pandas.DataFrame` Returns ------- :class:`pandas.DataFrame` """ data = self.to_json() if data['total'] == 0: raise DataNotFound('Your search matches no results') # res = {'date': [], 'lat': [], 'lon': [], 'wmo': [], 'status_name': [], 'status_code': []} # res = {'date': [], 'lat': [], 'lon': [], 'wmo': [], 'status_name': [], 'status_code': [], 'ship_name': []} res = {'date': [], 'lat': [], 'lon': [], 'wmo': [], 'status_name': [], 'status_code': [], 'program': [], 'country': [], 'model': []} # status = {'REGISTERED': None, 'OPERATIONAL': None, 'INACTIVE': None, 'CLOSED': None, # 'CONFIRMED': None, 'OPERATIONAL': None, 'PROBABLE': None, 'REGISTERED': None} for irow, ptf in enumerate(data['data']): # if irow == 0: # print(ptf) res['lat'].append(ptf['ptfDepl']['lat']) res['lon'].append(ptf['ptfDepl']['lon']) res['date'].append(ptf['ptfDepl']['deplDate']) res['wmo'].append(ptf['ref']) # res['wmo'].append(ptf['wmos'][-1]['wmo']) # res['wmo'].append(float_wmo(ptf['ref'])) # will not work for some CONFIRMED, PROBABLE or REGISTERED floats # res['wmo'].append(float_wmo(ptf['wmos'][-1]['wmo'])) res['status_code'].append(ptf['ptfStatus']['id']) res['status_name'].append(ptf['ptfStatus']['name']) # res['ship_name'].append(ptf['ptfDepl']['shipName']) program = ptf['program']['nameShort'].replace("_", " ") if ptf['program']['nameShort'] else ptf['program'][ 'nameShort'] res['program'].append(program) res['country'].append(ptf['program']['country']['nameShort']) res['model'].append(ptf['ptfModel']['nameShort']) # if status[ptf['ptfStatus']['name']] is None: # status[ptf['ptfStatus']['name']] = ptf['ptfStatus']['description'] df = pd.DataFrame(res) df = df.astype({'date': 'datetime64[s]'}) df = df.sort_values(by='date').reset_index(drop=True) # df = df[ (df['status_name'] == 'CLOSED') | (df['status_name'] == 'OPERATIONAL')] # Select only floats that have been deployed and returned data # print(status) return df
def plot_status(self, **kwargs ): """Quick plot of the deployment plan Named arguments are passed to :class:`plot.scatter_map` Returns ------- fig: :class:`matplotlib.figure.Figure` ax: :class:`matplotlib.axes.Axes` """ from .plot.plot import scatter_map df = self.to_dataframe() fig, ax = scatter_map(df, x='lon', y='lat', hue='status_code', traj=False, cmap='deployment_status', **kwargs) ax.set_title("Argo network deployment plan\n%s\nSource: OceanOPS API as of %s" % ( self.box_name, pd.to_datetime('now', utc=True).strftime("%Y-%m-%d %H:%M:%S")), fontsize=12 ) return fig, ax
@deprecated def cast_types(ds): # noqa: C901 """ Make sure variables are of the appropriate types according to Argo #todo: This is hard coded, but should be retrieved from an API somewhere. Should be able to handle all possible variables encountered in the Argo dataset. Parameter --------- :class:`xarray.DataSet` Returns ------- :class:`xarray.DataSet` """ list_str = [ "PLATFORM_NUMBER", "DATA_MODE", "DIRECTION", "DATA_CENTRE", "DATA_TYPE", "FORMAT_VERSION", "HANDBOOK_VERSION", "PROJECT_NAME", "PI_NAME", "STATION_PARAMETERS", "DATA_CENTER", "DC_REFERENCE", "DATA_STATE_INDICATOR", "PLATFORM_TYPE", "FIRMWARE_VERSION", "POSITIONING_SYSTEM", "PROFILE_PRES_QC", "PROFILE_PSAL_QC", "PROFILE_TEMP_QC", "PARAMETER", "SCIENTIFIC_CALIB_EQUATION", "SCIENTIFIC_CALIB_COEFFICIENT", "SCIENTIFIC_CALIB_COMMENT", "HISTORY_INSTITUTION", "HISTORY_STEP", "HISTORY_SOFTWARE", "HISTORY_SOFTWARE_RELEASE", "HISTORY_REFERENCE", "HISTORY_QCTEST", "HISTORY_ACTION", "HISTORY_PARAMETER", "VERTICAL_SAMPLING_SCHEME", "FLOAT_SERIAL_NO", "SOURCE", "EXPOCODE", "QCLEVEL", ] list_int = [ "PLATFORM_NUMBER", "WMO_INST_TYPE", "WMO_INST_TYPE", "CYCLE_NUMBER", "CONFIG_MISSION_NUMBER", ] list_datetime = [ "REFERENCE_DATE_TIME", "DATE_CREATION", "DATE_UPDATE", "JULD", "JULD_LOCATION", "SCIENTIFIC_CALIB_DATE", "HISTORY_DATE", "TIME" ] def fix_weird_bytes(x): x = x.replace(b"\xb1", b"+/-") return x fix_weird_bytes = np.vectorize(fix_weird_bytes) def cast_this(da, type): """ Low-level casting of DataArray values """ try: da.values = da.values.astype(type) da.attrs["casted"] = 1 except Exception: msg = "Oops! %s occurred. Fail to cast <%s> into %s for: %s. Encountered unique values: %s" % (sys.exc_info()[0], str(da.dtype), type, da.name, str(np.unique(da))) log.debug(msg) return da def cast_this_da(da): """ Cast any DataArray """ v = da.name da.attrs["casted"] = 0 if v in list_str and da.dtype == "O": # Object if v in ["SCIENTIFIC_CALIB_COEFFICIENT"]: da.values = fix_weird_bytes(da.values) da = cast_this(da, str) if v in list_int: # and da.dtype == 'O': # Object da = cast_this(da, np.int32) if v in list_datetime and da.dtype == "O": # Object if ( "conventions" in da.attrs and da.attrs["conventions"] == "YYYYMMDDHHMISS" ): if da.size != 0: if len(da.dims) <= 1: val = da.astype(str).values.astype("U14") # This should not happen, but still ! That's real world data val[val == " "] = "nan" da.values = pd.to_datetime(val, format="%Y%m%d%H%M%S") else: s = da.stack(dummy_index=da.dims) val = s.astype(str).values.astype("U14") # This should not happen, but still ! That's real world data val[val == ""] = "nan" val[val == " "] = "nan" # s.values = pd.to_datetime(val, format="%Y%m%d%H%M%S") da.values = s.unstack("dummy_index") da = cast_this(da, 'datetime64[s]') else: da = cast_this(da, 'datetime64[s]') elif v == "SCIENTIFIC_CALIB_DATE": da = cast_this(da, str) s = da.stack(dummy_index=da.dims) s.values = pd.to_datetime(s.values, format="%Y%m%d%H%M%S") da.values = (s.unstack("dummy_index")).values da = cast_this(da, 'datetime64[s]') if "QC" in v and "PROFILE" not in v and "QCTEST" not in v: if da.dtype == "O": # convert object to string da = cast_this(da, str) # Address weird string values: # (replace missing or nan values by a '0' that will be cast as an integer later if da.dtype == "<U3": # string, len 3 because of a 'nan' somewhere ii = ( da == " " ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) ii = ( da == "nan" ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) # Get back to regular U1 string da = cast_this(da, np.dtype("U1")) if da.dtype == "<U1": # string ii = ( da == "" ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) ii = ( da == " " ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) ii = ( da == "n" ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) # finally convert QC strings to integers: da = cast_this(da, np.int32) if da.dtype == 'O': # By default, try to cast as float: da = cast_this(da, np.float32) if da.dtype != "O": da.attrs["casted"] = 1 return da for v in ds.variables: try: ds[v] = cast_this_da(ds[v]) except Exception: print("Oops!", sys.exc_info()[0], "occurred.") print("Fail to cast: %s " % v) print("Encountered unique values:", np.unique(ds[v])) raise return ds def cast_Argo_variable_type(ds): """ Ensure that all dataset variables are of the appropriate types according to Argo references Parameter --------- :class:`xarray.DataSet` Returns ------- :class:`xarray.DataSet` """ list_str = [ "PLATFORM_NUMBER", "DATA_MODE", "DIRECTION", "DATA_CENTRE", "DATA_TYPE", "FORMAT_VERSION", "HANDBOOK_VERSION", "PROJECT_NAME", "PI_NAME", "STATION_PARAMETERS", "DATA_CENTER", "DC_REFERENCE", "DATA_STATE_INDICATOR", "PLATFORM_TYPE", "FIRMWARE_VERSION", "POSITIONING_SYSTEM", "PARAMETER", "SCIENTIFIC_CALIB_EQUATION", "SCIENTIFIC_CALIB_COEFFICIENT", "SCIENTIFIC_CALIB_COMMENT", "HISTORY_INSTITUTION", "HISTORY_STEP", "HISTORY_SOFTWARE", "HISTORY_SOFTWARE_RELEASE", "HISTORY_REFERENCE", "HISTORY_QCTEST", "HISTORY_ACTION", "HISTORY_PARAMETER", "VERTICAL_SAMPLING_SCHEME", "FLOAT_SERIAL_NO", "PARAMETER_DATA_MODE", # Trajectory file variables: 'TRAJECTORY_PARAMETERS', 'POSITION_ACCURACY', 'GROUNDED', 'SATELLITE_NAME', 'HISTORY_INDEX_DIMENSION', # Technical file variables: 'TECHNICAL_PARAMETER_NAME', 'TECHNICAL_PARAMETER_VALUE', 'PTT', # Metadata file variables: 'END_MISSION_STATUS', 'TRANS_SYSTEM', 'TRANS_SYSTEM_ID', 'TRANS_FREQUENCY', 'PLATFORM_FAMILY', 'PLATFORM_MAKER', 'MANUAL_VERSION', 'STANDARD_FORMAT_ID', 'DAC_FORMAT_ID', 'ANOMALY', 'BATTERY_TYPE', 'BATTERY_PACKS', 'CONTROLLER_BOARD_TYPE_PRIMARY', 'CONTROLLER_BOARD_TYPE_SECONDARY', 'CONTROLLER_BOARD_SERIAL_NO_PRIMARY', 'CONTROLLER_BOARD_SERIAL_NO_SECONDARY', 'SPECIAL_FEATURES', 'FLOAT_OWNER', 'OPERATING_INSTITUTION', 'CUSTOMISATION', 'DEPLOYMENT_PLATFORM', 'DEPLOYMENT_CRUISE_ID', 'DEPLOYMENT_REFERENCE_STATION_ID', 'LAUNCH_CONFIG_PARAMETER_NAME', 'CONFIG_PARAMETER_NAME', 'CONFIG_MISSION_COMMENT', 'SENSOR', 'SENSOR_MAKER', 'SENSOR_MODEL', 'SENSOR_SERIAL_NO', 'PARAMETER_SENSOR', 'PARAMETER_UNITS', 'PARAMETER_ACCURACY', 'PARAMETER_RESOLUTION', 'PREDEPLOYMENT_CALIB_EQUATION', 'PREDEPLOYMENT_CALIB_COEFFICIENT', 'PREDEPLOYMENT_CALIB_COMMENT', ] [list_str.append("PROFILE_{}_QC".format(v)) for v in list(ArgoNVSReferenceTables().tbl(3)["altLabel"])] list_int = [ "PLATFORM_NUMBER", "WMO_INST_TYPE", "WMO_INST_TYPE", "CYCLE_NUMBER", "CONFIG_MISSION_NUMBER", # Trajectory file variables: 'JULD_STATUS', 'JULD_ADJUSTED_STATUS', 'JULD_DESCENT_START_STATUS', 'JULD_FIRST_STABILIZATION_STATUS', 'JULD_DESCENT_END_STATUS', 'JULD_PARK_START_STATUS', 'JULD_PARK_END_STATUS', 'JULD_DEEP_DESCENT_END_STATUS', 'JULD_DEEP_PARK_START_STATUS', 'JULD_DEEP_ASCENT_START_STATUS', 'JULD_ASCENT_START_STATUS', 'JULD_ASCENT_END_STATUS', 'JULD_TRANSMISSION_START_STATUS', 'JULD_FIRST_MESSAGE_STATUS', 'JULD_FIRST_LOCATION_STATUS', 'JULD_LAST_LOCATION_STATUS', 'JULD_LAST_MESSAGE_STATUS', 'JULD_TRANSMISSION_END_STATUS', 'REPRESENTATIVE_PARK_PRESSURE_STATUS', ] list_datetime = [ "REFERENCE_DATE_TIME", "DATE_CREATION", "DATE_UPDATE", "JULD", "JULD_LOCATION", "SCIENTIFIC_CALIB_DATE", "HISTORY_DATE", "TIME", # Metadata file variables: 'LAUNCH_DATE', 'START_DATE', 'STARTUP_DATE', 'END_MISSION_DATE', ] def cast_this(da, type): """ Low-level casting of DataArray values """ try: # da.values = da.values.astype(type) da = da.astype(type) da.attrs["casted"] = 1 except Exception: print("Oops!", sys.exc_info()[0], "occurred.") print("Fail to cast %s[%s] from '%s' to %s" % (da.name, da.dims, da.dtype, type)) try: print("Unique values:", np.unique(da)) except: print("Can't read unique values !") pass return da def cast_this_da(da): """ Cast any DataArray """ # print("Casting %s ..." % da.name) da.attrs["casted"] = 0 if v in list_str and da.dtype == "O": # Object da = cast_this(da, str) if v in list_int: # and da.dtype == 'O': # Object if ( "conventions" in da.attrs and da.attrs["conventions"] in ["Argo reference table 19", "Argo reference table 21"] ): # Some values may be missing, and the _FillValue=" " cannot be casted as an integer. # so, we replace missing values with a 999: val = da.astype(str).values val[np.where(val == 'nan')] = '999' da.values = val da = cast_this(da, int) if v in list_datetime and da.dtype == "O": # Object if ( "conventions" in da.attrs and da.attrs["conventions"] == "YYYYMMDDHHMISS" ): if da.size != 0: if len(da.dims) <= 1: val = da.astype(str).values.astype("U14") # This should not happen, but still ! That's real world data val[val == " "] = "nan" da.values = pd.to_datetime(val, format="%Y%m%d%H%M%S") else: s = da.stack(dummy_index=da.dims) val = s.astype(str).values.astype("U14") # This should not happen, but still ! That's real world data val[val == " "] = "nan" s.values = pd.to_datetime(val, format="%Y%m%d%H%M%S") da.values = s.unstack("dummy_index") da = cast_this(da, 'datetime64[s]') else: da = cast_this(da, 'datetime64[s]') elif v == "SCIENTIFIC_CALIB_DATE": da = cast_this(da, str) s = da.stack(dummy_index=da.dims) s.values = pd.to_datetime(s.values, format="%Y%m%d%H%M%S") da.values = (s.unstack("dummy_index")).values da = cast_this(da, 'datetime64[s]') if "QC" in v and "PROFILE" not in v and "QCTEST" not in v: if da.dtype == "O": # convert object to string da = cast_this(da, str) # Address weird string values: # (replace missing or nan values by a '0' that will be cast as an integer later if da.dtype == "<U3": # string, len 3 because of a 'nan' somewhere ii = ( da == " " ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) ii = ( da == "nan" ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) # Get back to regular U1 string da = cast_this(da, np.dtype("U1")) if da.dtype == "<U1": # string ii = ( da == "" ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) ii = ( da == " " ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) ii = ( da == "n" ) # This should not happen, but still ! That's real world data da = xr.where(ii, "0", da) # finally convert QC strings to integers: da = cast_this(da, int) if da.dtype != "O": da.attrs["casted"] = 1 return da for v in ds.variables: try: ds[v] = cast_this_da(ds[v]) except Exception: print("Oops!", sys.exc_info()[0], "occurred.") print("Fail to cast: %s " % v) print("Encountered unique values:", np.unique(ds[v])) raise return ds def log_argopy_callerstack(level='debug'): """log the caller’s stack""" froot = str(pathlib.Path(__file__).parent.resolve()) for ideep, frame in enumerate(inspect.stack()[1:]): if os.path.join('argopy', 'argopy') in frame.filename: # msg = ["└─"] # [msg.append("─") for ii in range(ideep)] msg = [""] [msg.append(" ") for ii in range(ideep)] msg.append("└─ %s:%i -> %s" % (frame.filename.replace(froot, ''), frame.lineno, frame.function)) msg = "".join(msg) if level == "info": log.info(msg) elif level == "debug": log.debug(msg) elif level == "warning": log.warning(msg)
[docs]class ArgoDocs: """ADMT documentation helper class Examples -------- >>> ArgoDocs().list >>> ArgoDocs().search("CDOM") >>> ArgoDocs().search("CDOM", where='abstract') >>> ArgoDocs(35385) >>> ArgoDocs(35385).ris >>> ArgoDocs(35385).abstract >>> ArgoDocs(35385).show() >>> ArgoDocs(35385).open_pdf() >>> ArgoDocs(35385).open_pdf(page=12) """ _catalogue = [ { "categorie": "Argo data formats", "title": "Argo user's manual", "doi": "10.13155/29825", "id": 29825 }, { "categorie": "Quality control", "title": "Argo Quality Control Manual for CTD and Trajectory Data", "doi": "10.13155/33951", "id": 33951 }, { "categorie": "Quality control", "title": "Argo quality control manual for dissolved oxygen concentration", "doi": "10.13155/46542", "id": 46542 }, { "categorie": "Quality control", "title": "Argo quality control manual for biogeochemical data", "doi": "10.13155/40879", "id": 40879 }, { "categorie": "Quality control", "title": "BGC-Argo quality control manual for the Chlorophyll-A concentration", "doi": "10.13155/35385", "id": 35385 }, { "categorie": "Quality control", "title": "BGC-Argo quality control manual for nitrate concentration", "doi": "10.13155/84370", "id": 84370 }, { "categorie": "Quality control", "title": "Quality control for BGC-Argo radiometry", "doi": "10.13155/62466", "id": 62466 }, { "categorie": "Cookbooks", "title": "Argo DAC profile cookbook", "doi": "10.13155/41151", "id": 41151 }, { "categorie": "Cookbooks", "title": "Argo DAC trajectory cookbook", "doi": "10.13155/29824", "id": 29824 }, { "categorie": "Cookbooks", "title": "DMQC Cookbook for Core Argo parameters", "doi": "10.13155/78994", "id": 78994 }, { "categorie": "Cookbooks", "title": "Processing Argo oxygen data at the DAC level", "doi": "10.13155/39795", "id": 39795 }, { "categorie": "Cookbooks", "title": "Processing Bio-Argo particle backscattering at the DAC level", "doi": "10.13155/39459", "id": 39459 }, { "categorie": "Cookbooks", "title": "Processing BGC-Argo chlorophyll-A concentration at the DAC level", "doi": "10.13155/39468", "id": 39468 }, { "categorie": "Cookbooks", "title": "Processing Argo measurement timing information at the DAC level", "doi": "10.13155/47998", "id": 47998 }, { "categorie": "Cookbooks", "title": "Processing BGC-Argo CDOM concentration at the DAC level", "doi": "10.13155/54541", "id": 54541 }, { "categorie": "Cookbooks", "title": "Processing Bio-Argo nitrate concentration at the DAC Level", "doi": "10.13155/46121", "id": 46121 }, { "categorie": "Cookbooks", "title": "Processing BGC-Argo Radiometric data at the DAC level", "doi": "10.13155/51541", "id": 51541 }, { "categorie": "Cookbooks", "title": "Processing BGC-Argo pH data at the DAC level", "doi": "10.13155/57195", "id": 57195 }, { "categorie": "Cookbooks", "title": "Description of the Argo GDAC File Checks: Data Format and Consistency Checks", "doi": "10.13155/46120", "id": 46120 }, { "categorie": "Cookbooks", "title": "Description of the Argo GDAC File Merge Process", "doi": "10.13155/52154", "id": 52154 }, { "categorie": "Cookbooks", "title": "BGC-Argo synthetic profile file processing and format on Coriolis GDAC", "doi": "10.13155/55637", "id": 55637 }, { "categorie": "Cookbooks", "title": "Argo GDAC cookbook", "doi": "10.13155/46202", "id": 46202 } ] class RIS: """RIS file structure from TXT file""" def __init__(self, file=None, fs=None): self.record = None self.fs = fs if file: self.parse(file) def parse(self, file): """Parse input file""" with self.fs.open(file, 'r') as f: TXTlines = f.readlines() lines = [] # Eliminate blank lines for line in TXTlines: line = line.strip() if len(line) > 0: lines.append(line) TXTlines = lines # record = {} for line in TXTlines: # print("\n>", line) if len(line) > 2: if line[2] == " ": tag = line[0:2] field = line[3:] # print("ok", {tag: field}) record[tag] = [field] else: # print("-", line) record[tag].append(line) elif len(line) == 2: record[line] = [] # else: # print("*", line) for key in record.keys(): record[key] = "; ".join(record[key]) self.record = record
[docs] def __init__(self, docid=None): self.docid = None self._ris = None from .stores import httpstore self._fs = httpstore(cache=True, cachedir=OPTIONS['cachedir']) if isinstance(docid, int): if docid in [doc['id'] for doc in self._catalogue]: self.docid = docid else: raise ValueError("Unknow document id")
def __repr__(self): summary = ["<argopy.ArgoDocs>"] if self.docid is not None: doc = [doc for doc in self._catalogue if doc['id'] == self.docid][0] summary.append("Title: %s" % doc['title']) summary.append("DOI: %s" % doc['doi']) summary.append("url: https://dx.doi.org/%s" % doc['doi']) summary.append("last pdf: %s" % self.pdf) if 'AF' in self.ris: summary.append("Authors: %s" % self.ris['AF']) summary.append("Abstract: %s" % self.ris['AB']) else: summary.append("- %i documents with a DOI are available in the catalogue" % len(self._catalogue)) summary.append("- Use the method 'search' to find a document id") summary.append("- Use the property 'list' to check out the catalogue") return "\n".join(summary) @property def list(self): """List of all available documents as a :class:`pandas.DataFrame`""" return pd.DataFrame(self._catalogue) @property def js(self): """Internal json record for a document""" if self.docid is not None: return [doc for doc in self._catalogue if doc['id'] == self.docid][0] else: raise ValueError("Select a document first !") @property def ris(self): """RIS record of a document""" if self.docid is not None: if self._ris is None: # Fetch RIS metadata for this document: import re file = self._fs.fs.cat_file("https://dx.doi.org/%s" % self.js['doi']) x = re.search('<a target="_blank" href="(https?:\/\/([^"]*))"\s+([^>]*)rel="nofollow">TXT<\/a>', str(file)) export_txt_url = x[1] self._ris = self.RIS(export_txt_url, fs=self._fs).record return self._ris else: raise ValueError("Select a document first !") @property def abstract(self): """Abstract of a document""" if self.docid is not None: return self.ris['AB'] else: raise ValueError("Select a document first !") @property def pdf(self): """Link to the online pdf version of a document""" if self.docid is not None: return self.ris['UR'] else: raise ValueError("Select a document first !")
[docs] def show(self, height=800): """Insert document in pdf in a notebook cell Parameters ---------- height: int Height in pixels of the cell """ if self.docid is not None: from IPython.core.display import HTML return HTML( '<embed src="%s" type="application/pdf" width="100%%" height="%ipx" />' % (self.ris['UR'], height)) else: raise ValueError("Select a document first !")
[docs] def open_pdf(self, page=None): """Open document in new browser tab Parameters ---------- page: int, optional Open directly a specific page number """ if self.docid is not None: import webbrowser url = self.pdf url += '#view=FitV&pagemode=thumbs' if page: url += '&page=%i' % page webbrowser.open_new(url) else: raise ValueError("Select a document first !")
[docs] def search(self, txt, where='title'): """Search for string in all documents title or abstract Parameters ---------- txt: str where: str, default='title' Where to search, can be 'title' or 'abstract' Returns ------- list """ results = [] for doc in self.list.iterrows(): docid = doc[1]['id'] if where == 'title': if txt.lower() in ArgoDocs(docid).js['title'].lower(): results.append(docid) elif where == 'abstract': if txt.lower() in ArgoDocs(docid).abstract.lower(): results.append(docid) return results