"""
This module manage options of the package
# Like always, largely inspired by xarray code:
# https://github.com/pydata/xarray/blob/cafab46aac8f7a073a32ec5aa47e213a9810ed54/xarray/core/options.py
"""
import os
import numpy as np
from argopy.errors import OptionValueError, FtpPathError
import warnings
# Define option names as seen by users:
DATA_SOURCE = "src"
LOCAL_FTP = "local_ftp"
DATASET = "dataset"
DATA_CACHE = "cachedir"
USER_LEVEL = "mode"
API_TIMEOUT = "api_timeout"
TRUST_ENV = "trust_env" # Get proxies information from HTTP_PROXY / HTTPS_PROXY environment variables if the parameter is True (False by default).
# Get proxy credentials from ~/.netrc file if present.
# Define the list of available options and default values:
OPTIONS = {
DATA_SOURCE: "erddap",
LOCAL_FTP: None,
DATASET: "phy",
DATA_CACHE: os.path.expanduser(os.path.sep.join(["~", ".cache", "argopy"])),
USER_LEVEL: "standard",
API_TIMEOUT: 60,
TRUST_ENV: False
}
# Define the list of possible values
_DATA_SOURCE_LIST = frozenset(["erddap", "localftp", "argovis"])
_DATASET_LIST = frozenset(["phy", "bgc", "ref"])
_USER_LEVEL_LIST = frozenset(["standard", "expert"])
# Define how to validate options:
def _positive_integer(value):
return isinstance(value, int) and value > 0
def validate_ftp(this_path):
if this_path is not None:
return check_localftp(this_path, errors='raise')
else:
return False
_VALIDATORS = {
DATA_SOURCE: _DATA_SOURCE_LIST.__contains__,
LOCAL_FTP: validate_ftp,
DATASET: _DATASET_LIST.__contains__,
DATA_CACHE: os.path.exists,
USER_LEVEL: _USER_LEVEL_LIST.__contains__,
API_TIMEOUT: lambda x: isinstance(x, int) and x > 0,
TRUST_ENV: lambda x: isinstance(x, bool)
}
[docs]class set_options:
"""Set options for argopy.
List of options:
- `dataset`: Define the Dataset to work with.
Default: `phy`. Possible values: `phy`, `bgc` or `ref`.
- `src`: Source of fetched data.
Default: `erddap`. Possible values: `erddap`, `localftp`, `argovis`
- `local_ftp`: Absolute path to a local GDAC ftp copy.
Default: `.`
- `cachedir`: Absolute path to a local cache directory.
Default: `~/.cache/argopy`
- `mode`: User mode.
Default: `standard`. Possible values: `standard` or `expert`.
- `api_timeout`: Define the time out of internet requests to web API, in seconds.
Default: 60
You can use `set_options` either as a context manager:
>>> import argopy
>>> with argopy.set_options(src='localftp'):
>>> ds = argopy.DataFetcher().float(3901530).to_xarray()
Or to set global options:
>>> argopy.set_options(src='localftp')
"""
[docs] def __init__(self, **kwargs):
self.old = {}
for k, v in kwargs.items():
if k not in OPTIONS:
raise ValueError(
"argument name %r is not in the set of valid options %r"
% (k, set(OPTIONS))
)
if k in _VALIDATORS and not _VALIDATORS[k](v):
raise OptionValueError(f"option {k!r} given an invalid value: {v!r}")
self.old[k] = OPTIONS[k]
self._apply_update(kwargs)
def _apply_update(self, options_dict):
# for k, v in options_dict.items():
# if k in _SETTERS:
# _SETTERS[k](v)
OPTIONS.update(options_dict)
def __enter__(self):
return
def __exit__(self, type, value, traceback):
self._apply_update(self.old)
def check_localftp(path, errors: str = "ignore"):
""" Check if the path has the expected GDAC ftp structure
Check if the path is structured like:
.
└── dac
├── aoml
├── ...
├── coriolis
├── ...
├── meds
└── nmdis
Parameters
----------
path: str
Path name to check
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
"""
dacs = [
"aoml",
"bodc",
"coriolis",
"csio",
"csiro",
"incois",
"jma",
"kma",
"kordi",
"meds",
"nmdis",
]
# Case 1:
check1 = (
os.path.isdir(path)
and os.path.isdir(os.path.join(path, "dac"))
and np.any([os.path.isdir(os.path.join(path, "dac", dac)) for dac in dacs])
)
if check1:
return True
elif errors == "raise":
# This was possible up to v0.1.3:
check2 = os.path.isdir(path) and np.any(
[os.path.isdir(os.path.join(path, dac)) for dac in dacs]
)
if check2:
raise FtpPathError(
"This path is no longer GDAC compliant for argopy.\n"
"Please make sure you point toward a path with a 'dac' folder:\n%s"
% path
)
else:
raise FtpPathError("This path is not GDAC compliant (no `dac` folder with legitimate sub-folder):\n%s" % path)
elif errors == "warn":
# This was possible up to v0.1.3:
check2 = os.path.isdir(path) and np.any(
[os.path.isdir(os.path.join(path, dac)) for dac in dacs]
)
if check2:
warnings.warn(
"This path is no longer GDAC compliant for argopy. This will raise an error in the future.\n"
"Please make sure you point toward a path with a 'dac' folder:\n%s"
% path
)
return False
else:
warnings.warn("This path is not GDAC compliant:\n%s" % path)
return False
else:
return False