Source code for pycif.plugins.datastreams.backgrounds.tm5_background.read

import xarray as xr
import numpy as np
from netCDF4 import Dataset
import datetime
import pandas as pd
from .....utils.datastores.empty import init_empty
from .....utils.hdf5 import _hdf5_lock




def read(
    self,
    name,
    varnames,
    dates,
    files,
    interpol_flx=False,
    tracer=None,
    model=None,
    **kwargs
):
    """Read TM5-4DVAR background concentrations into a pyCIF datastore.

    Iterates over the unique files referenced in ``files``; for each new
    file, decodes the station-ID variable (``tracer.id_varname``) into
    station/network/agl (above-ground-level) components, reads the time
    stamps (``tracer.time_varname``) and the simulated-concentration array
    (``self.varname``), meshes stations against time stamps, and packs the
    result into an empty pyCIF datastore.

    Note:
        Only the datastore built from the last (in iteration order)
        distinct file in ``files`` is returned: ``ds`` is reassigned on
        each new file rather than accumulated/concatenated across files.

    Args:
        self: the background datastream Plugin.
        name: name of the observed parameter/component.
        varnames: unused (kept for interface consistency with other
            datastream ``read`` functions).
        dates: list of ``[start, end]`` date pairs, aligned with ``files``.
        files: list of TM5-4DVAR annual background files, aligned with
            ``dates``. Consecutive duplicate entries are only read once.
        interpol_flx (bool): unused here (kept for interface consistency).
        tracer: the background tracer, used for ``id_varname``,
            ``time_varname`` and the ``numscale`` scaling factor.
        model: unused here (kept for interface consistency).

    Returns:
        pandas.DataFrame: pyCIF datastore (as built by ``init_empty``) for
        the last file processed, with ``date``, ``station``, ``network``,
        ``parameter`` and ``duration`` metadata columns, and ``obserror``/
        ``spec`` main-data columns.
    """
    
    trcr_bc = []
    out_dates = []
    ref_file = ""
    
    for dd, ff in zip(dates, files):
        if ff != ref_file:
            ref_file = ff
            with _hdf5_lock:
                with Dataset(ff) as bkg:
                    # Station ids
                    ids = [''.join([c.decode().lower() for c in name])
                           for name in bkg[tracer.id_varname][:]]
                    ids = pd.Series(ids).str.split("_", expand=True)
                    stations = ids[0].values
                    network = ids[1].values
                    agl = ids[2].values
                    # Time steps
                    ts = np.array([
                        datetime.datetime(*t) for t in bkg[tracer.time_varname][:]
                    ])
                    # Simulations from TM5
                    data = bkg.variables[self.varname][:]
                mesh_stat, mesh_ts = np.meshgrid(stations, ts)
                mesh_network, mesh_ts = np.meshgrid(network, ts)
                mesh_agl, mesh_ts = np.meshgrid(agl, ts)
                ds = init_empty()
                ds[("metadata", "date")] = mesh_ts.flatten()
                ds[("metadata", "station")] = mesh_stat.flatten()
                ds[("metadata", "network")] = mesh_network.flatten()
                ds[("metadata", "parameter")] = name
                ds[("metadata", "duration")] = 1
                ds[("maindata", "obserror")] = 0
                ds[("maindata", "spec")] = \
                    data.flatten() * getattr(tracer, "numscale", 1)
    
    return ds