Skip to content
Snippets Groups Projects
Commit 7487ae48 authored by lukas leufen's avatar lukas leufen
Browse files

Added all docstrings, add example for hourly auth into join_settings. Will...

Added all docstrings, add example for hourly auth into join_settings. Will become untracked after next commit again.
parent 1cfba139
No related branches found
No related tags found
3 merge requests!125Release v0.10.0,!124Update Master to new version v0.10.0,!91WIP: Resolve "create sphinx docu"
Pipeline #35119 failed
"""Collection of configuration functions, paths and classes."""
from .path_config import ROOT_PATH, prepare_host, set_experiment_name, set_bootstrap_path, check_path_and_create from .path_config import ROOT_PATH, prepare_host, set_experiment_name, set_bootstrap_path, check_path_and_create
\ No newline at end of file
def join_settings(sampling="daily"): """Settings to access not public join data."""
from typing import Tuple, Dict
def join_settings(sampling="daily") -> Tuple[str, Dict]:
"""
Set url for join and required headers.
Headers information is not required for daily resolution. For hourly data "Authorization": "<yourtoken>" is required
to retrieve any data at all.
:param sampling: temporal resolution to access. Hourly data requires authorisation.
:return: Service url and optional headers
"""
if sampling == "daily": # pragma: no branch if sampling == "daily": # pragma: no branch
TOAR_SERVICE_URL = 'https://join.fz-juelich.de/services/rest/surfacedata/' TOAR_SERVICE_URL = 'https://join.fz-juelich.de/services/rest/surfacedata/'
headers = {} headers = {}
elif sampling == "hourly": elif sampling == "hourly":
TOAR_SERVICE_URL = 'https://join.fz-juelich.de/services/rest/surfacedata/' TOAR_SERVICE_URL = 'https://join.fz-juelich.de/services/rest/surfacedata/'
headers = {} headers = {"Authorization": "Token 12345"}
else: else:
raise NameError(f"Given sampling {sampling} is not supported, choose from either daily or hourly sampling.") raise NameError(f"Given sampling {sampling} is not supported, choose from either daily or hourly sampling.")
return TOAR_SERVICE_URL, headers return TOAR_SERVICE_URL, headers
"""Functions related to path and os name setting."""
import logging import logging
import os import os
import re import re
import socket import socket
from typing import Tuple
ROOT_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) ROOT_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
def prepare_host(create_new=True, sampling="daily"): def prepare_host(create_new=True, sampling="daily") -> str:
"""
Set up host path.
Warning: This functions can only handle known hosts. For the moment, please add your hostname hardcoded here. For
future, this will be replace by a more flexible configuration file setup.
:param create_new: Create new path if enabled
:param sampling: sampling rate to separate data physically by temporal resolution
:return: full path of data
"""
hostname = socket.gethostname() hostname = socket.gethostname()
runner_regex = re.compile(r"runner-.*-project-2411-concurrent-\d+") runner_regex = re.compile(r"runner-.*-project-2411-concurrent-\d+")
try: try:
...@@ -41,12 +54,25 @@ def prepare_host(create_new=True, sampling="daily"): ...@@ -41,12 +54,25 @@ def prepare_host(create_new=True, sampling="daily"):
return path return path
def set_experiment_name(experiment_date=None, experiment_path=None, sampling=None): def set_experiment_name(experiment_name=None, experiment_path=None, sampling=None) -> Tuple[str, str]:
if experiment_date is None: """
Set name of experiment and its path.
* Experiment name is set to `TestExperiment` if not provided in kwargs. If a name is given, this string is expanded
by suffix `_network`. Experiment name is always expanded by `_<sampling>` as ending suffix if sampling is given.
* Experiment path is set to `ROOT_PATH/<exp_name>` if not provided or otherwise use `<experiment_path>/<exp_name>`
:param experiment_name: custom experiment name
:param experiment_path: custom experiment path
:param sampling: sampling rate as string to add to experiment name
:return: experiment name and full experiment path
"""
if experiment_name is None:
experiment_name = "TestExperiment" experiment_name = "TestExperiment"
else: else:
experiment_name = f"{experiment_date}_network" experiment_name = f"{experiment_name}_network"
if sampling == "hourly": if sampling is not None:
experiment_name += f"_{sampling}" experiment_name += f"_{sampling}"
if experiment_path is None: if experiment_path is None:
experiment_path = os.path.abspath(os.path.join(ROOT_PATH, experiment_name)) experiment_path = os.path.abspath(os.path.join(ROOT_PATH, experiment_name))
...@@ -55,14 +81,30 @@ def set_experiment_name(experiment_date=None, experiment_path=None, sampling=Non ...@@ -55,14 +81,30 @@ def set_experiment_name(experiment_date=None, experiment_path=None, sampling=Non
return experiment_name, experiment_path return experiment_name, experiment_path
def set_bootstrap_path(bootstrap_path, data_path, sampling): def set_bootstrap_path(bootstrap_path: str, data_path: str, sampling: str) -> str:
"""
Set path for bootstrap input data.
Either use given bootstrap_path or create additional folder in same directory like data path.
:param bootstrap_path: custom path to store bootstrap data
:param data_path: path of data for default bootstrap path
:param sampling: sampling rate to add, if path is set to default
:return: full bootstrap path
"""
if bootstrap_path is None: if bootstrap_path is None:
bootstrap_path = os.path.join(data_path, "..", f"bootstrap_{sampling}") bootstrap_path = os.path.join(data_path, "..", f"bootstrap_{sampling}")
check_path_and_create(bootstrap_path) check_path_and_create(bootstrap_path)
return bootstrap_path return os.path.abspath(bootstrap_path)
def check_path_and_create(path: str) -> None:
"""
Check a given path and create if not existing.
def check_path_and_create(path): :param path: path to check and create
"""
try: try:
os.makedirs(path) os.makedirs(path)
logging.debug(f"Created path: {path}") logging.debug(f"Created path: {path}")
......
...@@ -29,6 +29,7 @@ def dict_to_xarray(d: Dict, coordinate_name: str) -> xr.DataArray: ...@@ -29,6 +29,7 @@ def dict_to_xarray(d: Dict, coordinate_name: str) -> xr.DataArray:
:param d: dictionary with 2D-xarrays :param d: dictionary with 2D-xarrays
:param coordinate_name: name of the new created axis (2D -> 3D) :param coordinate_name: name of the new created axis (2D -> 3D)
:return: combined xarray :return: combined xarray
""" """
xarray = None xarray = None
...@@ -51,6 +52,7 @@ def float_round(number: float, decimals: int = 0, round_type: Callable = math.ce ...@@ -51,6 +52,7 @@ def float_round(number: float, decimals: int = 0, round_type: Callable = math.ce
:param decimals: numbers of decimals of the rounding operations (default 0 -> round to next integer value) :param decimals: numbers of decimals of the rounding operations (default 0 -> round to next integer value)
:param round_type: the actual rounding operation. Can be any callable function like math.ceil, math.floor or python :param round_type: the actual rounding operation. Can be any callable function like math.ceil, math.floor or python
built-in round operation. built-in round operation.
:return: rounded number with desired precision :return: rounded number with desired precision
""" """
multiplier = 10. ** decimals multiplier = 10. ** decimals
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment