Source code for pycif.plugins.models.iconart.io.inputs.obs

import pandas as pd
import numpy as np
from ......utils.datastores.empty import init_empty
from ......utils.check.errclass import CifError




def make_obs(self, ddi, datastore, runsubdir, mode, tracer, input_type, do_simu=True):
    """Dumps observation locations and time steps to obs.txt to speed up 
    ICON output post-processing
    """

    # If empty datastore, do nothing
    if datastore.size == 0:
        return

    if self.reset_obs[ddi]:
        self.reset_obs[ddi] = False

    # If do not need to do ICON-ART simulation, just update iniobs
    if not do_simu:
        self.iniobs[ddi] = True
        return

    # Skip if input type is not "concs", i.e.,
    # auxiliary data such as pressure
    if input_type != "concs":
        return

    # Only once for perturbed tracer
    if "__sample#" in tracer:
        sample_id = tracer.split("__sample#")[1]
        if int(sample_id) > 0:
            return

    # Include only part of the datastore
    metadata = datastore["metadata"]

    # Fetch all cell indexes
    maincols = ['station', 'network', 'i',
                'lon', 'lat', 'alt', 'tstep', 'parameter']
    data = metadata.loc[:, maincols + ['level']]
    if not self.full_interpolation:
        data.loc[:, 'index'] = metadata.index
        data = data.groupby(['index', 'tstep']).first()
        data = data.reset_index()
        data.index = data['index']
        data.loc[data.index[0], 'level'] = 0
        data.loc[data.index[1], 'level'] = \
            self.domain.nlev - 1

    # change parameter in ref_parameter
    data.loc[:, 'parameter'] = \
        data.apply(
            lambda row: row['parameter'].split('__sample#')[0].upper(),
            axis=1
    )

    # Convert CIF level to ICON level
    data['level'] = (self.domain.nlev - 1) - data['level']
    data = data.rename(columns={'level': 'icon_ilevel'})

    # Update csv file if necessary
    obs_file = f"{runsubdir}/obs.csv"
    if self.iniobs[ddi]:
        data_orig = pd.read_csv(obs_file,
                                index_col=0,
                                na_values=-999)
        data = pd.concat([data_orig, data], axis=0)

    # Check if vertical information is given by alt or level
    alt_obs = data['alt'].values
    ilevels_obs = data['icon_ilevel'].values
    mask_nan_alt = ~pd.notnull(alt_obs)
    num_both_nans = np.sum(~pd.notnull(ilevels_obs[mask_nan_alt]))
    if num_both_nans:
        raise CifError("Vertical information about "
                        "some observations is missing...")

    # Write in a csv file
    # data.index = range(len(data))
    data.to_csv(obs_file, na_rep='-999')

    # Keep in memory that observations were already dumped for that period
    self.iniobs[ddi] = True