Source code for pycif.plugins.datastreams.backgrounds.carboscope_bg.fetch

import datetime
import glob
import os
from logging import debug, info

import numpy as np
import pandas as pd
from netCDF4 import Dataset

from .....utils import path
from .....utils.dates import date_range
from .....utils.check.errclass import CifError




def fetch(ref_dir, ref_file, date_interval, target_dir, tracer=None, **kwargs):
    """Locate CARBOSCOPE background files and link them to the working directory.

    The requested date interval is expanded to cover full calendar years
    (CARBOSCOPE station files are not split by period), all files matching
    ``ref_dir/ref_file`` are symlinked into ``target_dir``, and hourly
    sub-periods spanning the expanded interval are generated.

    Args:
        ref_dir: directory holding the CARBOSCOPE background files.
        ref_file: glob pattern (relative to ``ref_dir``) matching the
            per-station files to fetch.
        date_interval: 2-element sequence ``(datei, datef)`` giving the
            requested date range.
        target_dir: directory where matching files are symlinked.
        tracer: the background datastream Plugin (unused here beyond being
            accepted for interface consistency).

    Returns:
        tuple: ``(list_files, list_dates)``, each a dict with a single key
        (the start of the expanded interval). ``list_files`` maps it to the
        list of linked file paths; ``list_dates`` maps it to the list of
        hourly ``[start, start + 1h]`` date pairs spanning the interval.

    Raises:
        CifError: if no file matches ``ref_dir/ref_file``.
    """

    # Reshape input interval to include full years
    datei, datef = date_interval
    datei = datetime.datetime(year=datei.year, month=1, day=1)
    datef = datetime.datetime(year=datef.year + 1, month=1, day=1)

    # Check that there are files
    list_available = glob.glob(f"{ref_dir}/{ref_file}")
    if list_available == []:
        raise CifError(
            "WARNING: No background files found for CARBOSCOPE in "
            f"{ref_dir}/{ref_file}"
        )

    debug(
        "Available files for CARBOSCOPE background: \n"
        + "\n".join([f"    - {f}" for f in list_available])
    )

    # Link files to datavect
    list_files = {datei: []}
    for f in list_available:
        target_file = f"{target_dir}/{os.path.basename(f)}"
        path.link(f, target_file)
        list_files[datei].append(target_file)

    list_dates = {
        datei: [
            [ddi, ddi + pd.Timedelta("1h")]
            for ddi in date_range(datei, datef, period="1h")
        ]
    }

    return list_files, list_dates