Source code for pycif.plugins.transforms.basic.vertical_interpolation.utils.sparse.vcoordfromfile

import numpy as np
import pandas as pd
from logging import debug
from .......utils.check.errclass import CifError




def vcoordfromfile(datastore, file_lev, **kwargs):
    """Computes vertical levels from pre-defined file

    Args:
        datastore (dict): dictionary of pd.Dataframes containing measurement
                          informations
        file_lev (str): path to the file defining station levels

    """

    lev_stats = \
        np.genfromtxt(file_lev, skip_header=1,
                      usecols=0, dtype=str).flatten()
    lev_infos = \
        np.genfromtxt(file_lev, skip_header=1,
                      usecols=3, dtype=int).flatten()

    # If duplicate stations, raise exception
    if np.unique(lev_stats).size < lev_stats.size:
        duplicates = np.unique(lev_stats, return_counts=True)
        raise CifError(
            f"There are duplicate stations defined in {file_lev}. Please check your file. \nDuplicate stations: {duplicates[0][duplicates[1] != 1]}")

    levels = pd.Series(index=datastore.index, dtype=int)
    ds_stations = datastore["metadata"]["station"].unique().astype(str)

    debug(f"Computing levels from static levels: \nfile_lev: {file_lev}\nstation infos: {lev_stats}\nstation in datastore: {ds_stations}")

    computed_level = 0
    computed_stations = []
    for s, linfo in zip(lev_stats, lev_infos):
        mask = datastore["metadata"]["station"].str.lower() == s.lower()
        if s.lower() in np.char.lower(ds_stations):
            computed_stations.append(s)

        levels.loc[mask] = linfo
        computed_level += mask.sum()

    if len(computed_stations) != len(ds_stations):
        missing = [s for s in ds_stations if s not in computed_stations]
        raise CifError(f"I could not compute levels for all data points. Please check the level-defining file ({file_lev})\nMissing stations in file_lev: {missing}")

    return levels