Skip to content

Commit

Permalink
Define path.py module
Browse files Browse the repository at this point in the history
  • Loading branch information
@ghiggi
ghiggi committed Nov 10, 2023
1 parent 7afc399 commit 16abdbe
Show file tree
Hide file tree
Showing 25 changed files with 795 additions and 599 deletions.
2 changes: 1 addition & 1 deletion disdrodb/api/configs.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import os

from disdrodb.api.checks import check_product, check_sensor_name
from disdrodb.api.io import define_config_dir
from disdrodb.api.path import define_config_dir
from disdrodb.utils.yaml import read_yaml

logger = logging.getLogger(__name__)
Expand Down
247 changes: 2 additions & 245 deletions disdrodb/api/io.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,252 +23,9 @@
import numpy as np

from disdrodb.api.checks import check_product
from disdrodb.api.path import get_disdrodb_path
from disdrodb.configs import get_base_dir
from disdrodb.utils.directories import check_directory_exists, count_files, list_directories, list_files


def get_disdrodb_path(
base_dir,
product,
data_source="",
campaign_name="",
check_exists=True,
):
"""Return the directory in the DISDRODB infrastructure.
If data_source and campaign_name are not specified it return the product directory.
If data_source is specified, it returns the data_source directory.
If campaign_source is specified, it returns the campaign_name directory.
Parameters
----------
base_dir : str
The disdrodb base directory
product : str
The DISDRODB product. It can be "RAW", "L0A", or "L0B".
data_source : str, optional
The data source. Must be specified if campaign_name is specified.
campaign_name : str, optional
The campaign_name.
check_exists : bool, optional
Whether to check if the directory exists. By default True.
Returns
-------
dir_path : str
Directory path
"""
from disdrodb.api.checks import check_base_dir
from disdrodb.utils.directories import check_directory_exists

# Check base_dir validity
base_dir = check_base_dir(base_dir)
if len(campaign_name) > 0:
if len(data_source) == 0:
raise ValueError("If campaign_name is specified, data_source must be specified.")

# Get directory
if product.upper() == "RAW":
dir_path = os.path.join(base_dir, "Raw", data_source, campaign_name)
else:
dir_path = os.path.join(base_dir, "Processed", data_source, campaign_name)
if check_exists:
check_directory_exists(dir_path)
return dir_path


def define_campaign_dir(
product,
data_source,
campaign_name,
base_dir=None,
check_exists=False,
):
"""Return the campaign directory in the DISDRODB infrastructure.
Parameters
----------
product : str
The DISDRODB product. It can be "RAW", "L0A", or "L0B".
data_source : str
The data source. Must be specified if campaign_name is specified.
campaign_name : str
The campaign_name.
base_dir : str, optional
The base directory of DISDRODB, expected in the format ``<...>/DISDRODB``.
If not specified, the path specified in the DISDRODB active configuration will be used.
check_exists : bool, optional
Whether to check if the directory exists. By default False.
Returns
-------
station_dir : str
Station data directory path
"""
base_dir = get_base_dir(base_dir)
campaign_dir = get_disdrodb_path(
base_dir=base_dir,
product=product,
data_source=data_source,
campaign_name=campaign_name,
check_exists=check_exists,
)
return str(campaign_dir)


def define_station_dir(
product,
data_source,
campaign_name,
station_name,
base_dir=None,
check_exists=False,
):
"""Return the station data directory in the DISDRODB infrastructure.
Parameters
----------
product : str
The DISDRODB product. It can be "RAW", "L0A", or "L0B".
data_source : str
The data source.
campaign_name : str
The campaign name.
station_name : str
The station name.
base_dir : str, optional
The base directory of DISDRODB, expected in the format ``<...>/DISDRODB``.
If not specified, the path specified in the DISDRODB active configuration will be used.
check_exists : bool, optional
Whether to check if the directory exists. By default False.
Returns
-------
station_dir : str
Station data directory path
"""
base_dir = get_base_dir(base_dir)
campaign_dir = get_disdrodb_path(
base_dir=base_dir,
product=product,
data_source=data_source,
campaign_name=campaign_name,
check_exists=check_exists,
)
if product.upper() == "RAW":
station_dir = os.path.join(campaign_dir, "data", station_name)
else:
station_dir = os.path.join(campaign_dir, product, station_name)
if check_exists:
check_directory_exists(station_dir)
return str(station_dir)


def define_metadata_dir(
product,
data_source,
campaign_name,
base_dir=None,
check_exists=False,
):
"""Return the metadata directory in the DISDRODB infrastructure.
Parameters
----------
product : str
The DISDRODB product. It can be "RAW", "L0A", or "L0B".
data_source : str
The data source.
campaign_name : str
The campaign name.
base_dir : str, optional
The base directory of DISDRODB, expected in the format ``<...>/DISDRODB``.
If not specified, the path specified in the DISDRODB active configuration will be used.
check_exists : bool, optional
Whether to check if the directory exists. By default False.
Returns
-------
metadata_dir : str
Station data directory path
"""
base_dir = get_base_dir(base_dir)
campaign_dir = define_campaign_dir(
base_dir=base_dir,
product=product,
data_source=data_source,
campaign_name=campaign_name,
check_exists=check_exists,
)
if product.upper() == "RAW":
metadata_dir = os.path.join(campaign_dir, "metadata")
else:
metadata_dir = os.path.join(campaign_dir, "metadata")
if check_exists:
check_directory_exists(metadata_dir)
return str(metadata_dir)


def define_metadata_filepath(
product,
data_source,
campaign_name,
station_name,
base_dir=None,
check_exists=False,
):
"""Return the station metadata filepath in the DISDRODB infrastructure.
Parameters
----------
product : str
The DISDRODB product. It can be "RAW", "L0A", or "L0B".
data_source : str
The data source.
campaign_name : str
The campaign name.
station_name : str
The station name.
base_dir : str, optional
The base directory of DISDRODB, expected in the format ``<...>/DISDRODB``.
If not specified, the path specified in the DISDRODB active configuration will be used.
check_exists : bool, optional
Whether to check if the directory exists. By default False.
Returns
-------
metadata_dir : str
Station data directory path
"""
base_dir = get_base_dir(base_dir)
metadata_dir = define_metadata_dir(
base_dir=base_dir,
product=product,
data_source=data_source,
campaign_name=campaign_name,
check_exists=False,
)
metadata_filepath = os.path.join(metadata_dir, f"{station_name}.yml")

if check_exists and not os.path.exists(metadata_filepath):
raise ValueError(f"The metadata file for {station_name} at {metadata_filepath} does not exists.")

return str(metadata_filepath)


def define_config_dir(product):
"""Define the config directory path of a given DISDRODB product."""
from disdrodb import __root_path__

if product.upper() in ["RAW", "L0A", "L0B"]:
dir_name = "l0"
else:
raise NotImplementedError(f"Product {product} not implemented.")
config_dir = os.path.join(__root_path__, "disdrodb", dir_name, "configs")
return config_dir


####---------------------------------------------------------------------------.
from disdrodb.utils.directories import count_files, list_directories, list_files


def _get_list_stations_dirs(product, campaign_dir):
Expand Down
Loading

0 comments on commit 16abdbe

Please sign in to comment.