API reference

earthcarekit.utils

Collection of basic utility functions.

Notes

This module depends on other internal modules:

• earthcarekit.typing

---

get_config

get_config(c: str | ECKConfig | None = None) -> ECKConfig

Returns the default or a given earthcarekit config object.

Parameters:

Name	Type	Description	Default
c	str | ECKConfig | None	A path to a config file (.toml) or None. If None, returns the default config. Defaults to None.	None

Returns:

Name	Type	Description
ECKConfig	ECKConfig	A config object.

Source code in earthcarekit/utils/_config.py

def get_config(c: str | ECKConfig | None = None) -> ECKConfig:
    """
    Returns the default or a given earthcarekit config object.

    Args:
        c (str | ECKConfig | None, optional): A path to a config file (.toml) or None. If None, returns the default config. Defaults to None.

    Returns:
        ECKConfig: A config object.
    """
    _config: ECKConfig
    if c is None:
        _config = read_config()
    elif isinstance(c, str):
        _config = read_config(c)
    elif isinstance(c, ECKConfig):
        _config = c
    else:
        raise TypeError(
            f"Invalid config! Either give a path to a eckit config TOML file or pass a instance of the class '{ECKConfig.__name__}'"
        )
    return _config

get_file_info_from_str

get_file_info_from_str(s: str) -> dict

Extract meta data from a EarthCARE product/file name.

Parameters:

Name	Type	Description	Default
s	str	The input string.	required

Raises:

Type	Description
ValueError	If the pattern of s does not match the EarthCARE product naming convention.

Returns:

Name	Type	Description
dict	dict	File meta data as a dict with the following keys: mission_id agency latency baseline file_type start_sensing_time start_processing_time orbit_number frame_id orbit_and_frame filename

Source code in earthcarekit/utils/parse/filename.py

def get_file_info_from_str(s: str) -> dict:
    """Extract meta data from a EarthCARE product/file name.

    Args:
        s (str): The input string.

    Raises:
        ValueError: If the pattern of `s` does not match the EarthCARE product naming convention.

    Returns:
        dict: File meta data as a dict with the following keys:

            - `mission_id`
            - `agency`
            - `latency`
            - `baseline`
            - `file_type`
            - `start_sensing_time`
            - `start_processing_time`
            - `orbit_number`
            - `frame_id`
            - `orbit_and_frame`
            - `filename`
    """
    m = re.match(FILE_INFO_REGEX, s)
    if m is None:
        raise ValueError(f"no match found; could not get a EarthCARE file info from string '{s}'")
    fn = m.group()[-60:]  # type: ignore
    (agy, lty, bl, ft, sst, pst, oaf) = m.groups()  # type: ignore
    return dict(
        mission_id="ECA",
        agency=agy,
        latency=lty,
        baseline=bl,
        file_type=ft,
        start_sensing_time=pd.Timestamp(sst),
        start_processing_time=pd.Timestamp(pst),
        orbit_number=int(oaf[:-1]),
        frame_id=oaf[-1],
        orbit_and_frame=oaf,
        filename=fn,
    )

get_maap_access_token

get_maap_access_token(token: str | None = None, config: str | ECKConfig | None = None) -> str

Retrieves MAAP access token from generated offline token

Parameters:

Name	Type	Description	Default
token	str | None	A temporary/offline ESA MAAP access token (to generate it visit: https://portal.maap.eo.esa.int/ini/services/auth/token/). Defaults to None.	None
config	str | ECKConfig | None	A path to a config file (.toml) or None. If None, returns the default config. Defaults to None.	None

Returns:

Name	Type	Description
str	str	Long-lasting ESA MAAP access token.

Source code in earthcarekit/utils/maap.py

def get_maap_access_token(
    token: str | None = None,
    config: str | ECKConfig | None = None,
) -> str:
    """Retrieves MAAP access token from generated offline token

    Args:
        token (str | None):
            A temporary/offline ESA MAAP access token (to generate it visit:
            https://portal.maap.eo.esa.int/ini/services/auth/token/). Defaults to None.
        config (str | ECKConfig | None): A path to a config file (.toml) or None. If None, returns
            the default config. Defaults to None.

    Returns:
        str: Long-lasting ESA MAAP access token.
    """
    return _get_maap_access_token(token or get_config(config).maap_token)

has_param

has_param(cls: Type[Any], param: str) -> bool

Checks if class has parameter in its signature.

Source code in earthcarekit/utils/_inspect.py

def has_param(cls: Type[Any], param: str) -> bool:
    """Checks if class has parameter in its signature."""
    return param in inspect.signature(cls).parameters

read_config

read_config(config_filepath: str | None = None) -> ECKConfig

Reads and return earthcare-kit configurations.

Source code in earthcarekit/utils/_config.py

def read_config(config_filepath: str | None = None) -> ECKConfig:
    """Reads and return earthcare-kit configurations."""
    if isinstance(config_filepath, str):
        config_filepath = os.path.abspath(config_filepath)
    elif config_filepath is None:
        config_filepath = get_default_config_filepath()
    else:
        raise TypeError(
            f"got invalid type for `path_to_config` ('{type(config_filepath).__name__}'), expected type 'str'"
        )

    if os.path.exists(config_filepath):
        text = Path(config_filepath).read_text(encoding="utf-8")
        # with open(config_filepath, "rb") as f:
        for field in ["data_directory", "image_directory"]:
            match = re.match(rf"^\[local\].*{field}\s*=\s*(\".*?\")", text, re.S)
            if match:
                for g in match.groups():
                    text = re.sub(r"\\+", "/", text).replace(g, "'" + g.strip('"') + "'")
        # config = tomllib.load(f)
        # print(text)
        config = tomllib.loads(text)
        try:
            if "Local_file_system" in config:
                data_dirpath = config["Local_file_system"]["data_directory"]
                image_dirpath = config["Local_file_system"]["image_directory"]
            else:
                data_dirpath = config["local"]["data_directory"]
                image_dirpath = config["local"]["image_directory"]

            download_backend: str
            user_type: Literal["commissioning", "calval", "open", "none"] = "none"
            collections: str | list[str] | None
            if "OADS_credentials" in config:
                oads_username = config.get("OADS_credentials", dict()).get("username", "")
                oads_password = config.get("OADS_credentials", dict()).get("password", "")
                collections = config.get("OADS_credentials", dict()).get("collections", None)
                download_backend = config.get("OADS_credentials", dict()).get("platform", "maap")
                maap_token = config.get("OADS_credentials", dict()).get("maap_token", "")
            else:
                oads_username = config.get("download", dict()).get("oads_username", "")
                oads_password = config.get("download", dict()).get("oads_password", "")
                collections = config.get("download", dict()).get("collections", None)
                download_backend = config.get("download", dict()).get("platform", "maap")
                maap_token = config.get("download", dict()).get("maap_token", "")
                maap_include_header_file = config.get("download", dict()).get(
                    "maap_include_header_file", True
                )

            if isinstance(collections, str):
                if collections.lower() == "commissioning":
                    user_type = "commissioning"
                    collections = get_collections_from_user_type_str(user_type)
                elif collections.lower() == "calval":
                    user_type = "calval"
                    collections = get_collections_from_user_type_str(user_type)
                elif collections.lower() == "open":
                    user_type = "open"
                    collections = get_collections_from_user_type_str(user_type)

            _collections: list[DisseminationCollection] = []
            if isinstance(collections, list):
                _collections = [DisseminationCollection(c) for c in collections]

            data_directory_structure = config.get("local", dict()).get(
                "data_directory_structure", dict()
            )
            subdir_template = data_directory_structure.get(
                "subdir_template",
                "{level}/{file_type}/{year}/{month}/{day}/{baseline}",
            )
            subdir_name_auxiliary_files = data_directory_structure.get(
                "subdir_name_auxiliary_files", "auxiliary_files"
            )
            subdir_name_orbit_files = data_directory_structure.get(
                "subdir_name_orbit_files", "orbit_files"
            )
            subdir_name_level0 = data_directory_structure.get("subdir_name_level0", "level0")
            subdir_name_level1b = data_directory_structure.get("subdir_name_level1b", "level1b")
            subdir_name_level1c = data_directory_structure.get("subdir_name_level1c", "level1c")
            subdir_name_level2a = data_directory_structure.get("subdir_name_level2a", "level2a")
            subdir_name_level2b = data_directory_structure.get("subdir_name_level2b", "level2b")

            eckit_config = ECKConfig(
                filepath=config_filepath,
                path_to_data=data_dirpath,
                path_to_images=image_dirpath,
                oads_username=oads_username,
                oads_password=oads_password,
                collections=_collections,
                maap_token=maap_token,
                maap_include_header_file=maap_include_header_file,
                download_backend=download_backend.lower(),
                user_type=user_type,
                subdir_template=subdir_template,
                subdir_name_auxiliary_files=subdir_name_auxiliary_files,
                subdir_name_orbit_files=subdir_name_orbit_files,
                subdir_name_level0=subdir_name_level0,
                subdir_name_level1b=subdir_name_level1b,
                subdir_name_level1c=subdir_name_level1c,
                subdir_name_level2a=subdir_name_level2a,
                subdir_name_level2b=subdir_name_level2b,
            )
            return eckit_config
        except AttributeError as e:
            raise AttributeError(f"Invalid config file is missing variable: {e}")

    raise FileNotFoundError(
        f"Missing config.toml file ({config_filepath})\n{DEFAULT_CONFIG_SETUP_INSTRUCTIONS}"
    )

search_files_by_regex

search_files_by_regex(root: PathLike, pattern: str) -> list[str]

Recursively searches for files in a directory that match a given regex pattern.

Parameters:

Name	Type	Description	Default
root	str	The root directory to start the search from.	required
pattern	str	A regular expression pattern to match file names against.	required

Return

list[str]: A list of absolute file paths that point to files with matching names.

Raises:

Type	Description
FileNotFoundError	If the root directory does not exist.
error	If the given pattern is not a valid regular expression.

Source code in earthcarekit/utils/path.py

def search_files_by_regex(root: PathLike, pattern: str) -> list[str]:
    """Recursively searches for files in a directory that match a given regex pattern.

    Args:
        root (str): The root directory to start the search from.
        pattern (str): A regular expression pattern to match file names against.

    Return:
        list[str]: A list of absolute file paths that point to files with matching names.

    Raises:
        FileNotFoundError: If the root directory does not exist.
        re.error: If the given pattern is not a valid regular expression.
    """
    root = str(root)
    if not os.path.exists(root):
        raise FileNotFoundError(
            f"{search_files_by_regex.__name__}() Root directory does not exist: {root}"
        )

    filepaths = []
    for dirpath, _, filenames in os.walk(root):
        for filename in filenames:
            filepath = os.path.join(dirpath, filename)
            if re.search(pattern, filename):
                filepaths.append(filepath)
    return filepaths

set_config

set_config(c: str | ECKConfig, verbose: bool = True) -> None

Creates or updates the default earthcarekit configuration file.

Parameters:

Name	Type	Description	Default
c	str | ECKConfig	Filepath to a configuration file (.toml) or configuration object.	required
verbose	bool	If True, prints a message to the console. Defaults to True.	True

Source code in earthcarekit/utils/_config.py

def set_config(c: str | ECKConfig, verbose: bool = True) -> None:
    """
    Creates or updates the default earthcarekit configuration file.

    Args:
        c (str | ECKConfig): Filepath to a configuration file (.toml) or configuration object.
        verbose (bool): If True, prints a message to the console. Defaults to True.
    """
    _set_config(c=c, verbose=verbose)

set_config_maap_token

set_config_maap_token(token: str) -> None

Updates the ESA MAAP access token in the default earthcarekit configuration file.

Parameters:

Name	Type	Description	Default
token	str	A temporary ESA MAAP access token (to generate it visit: https://portal.maap.eo.esa.int/ini/services/auth/token/).	required

Source code in earthcarekit/utils/_config.py

def set_config_maap_token(token: str) -> None:
    """
    Updates the ESA MAAP access token in the default earthcarekit configuration file.

    Args:
        token (str): A temporary ESA MAAP access token (to generate it visit: https://portal.maap.eo.esa.int/ini/services/auth/token/).
    """
    _config: ECKConfig = read_config()
    _config.maap_token = token
    _set_config(
        _config,
        alt_msg="Set MAAP access token",
    )

set_config_to_maap

set_config_to_maap() -> None

Sets the download backend to the ESA MAAP system in the default earthcarekit configuration file.

Source code in earthcarekit/utils/_config.py

def set_config_to_maap() -> None:
    """Sets the download backend to the ESA MAAP system in the default earthcarekit configuration file."""
    _config: ECKConfig = read_config()
    _config.download_backend = "maap"
    _set_config(
        _config,
        alt_msg=f"Set download backend to {_config.download_backend.upper()}",
    )