Source code for pycif.plugins.datastreams.fluxes.point_sources.fetch

import datetime
import glob
import os
from logging import debug

import numpy as np
import pandas as pd

from .....utils import path




def fetch(
    ref_dir, ref_file, input_interval, target_dir, tracer=None, component=None, **kwargs
):
    """Fetch the point-source CSV file and build the hourly date intervals it covers.

    The same CSV file (holding all point sources and their validity
    periods) is linked once into `target_dir` and reused for every hourly
    sub-interval of `input_interval`; per-source filtering by validity
    period is done later in `read`.

    Args:
        ref_dir (str): Directory containing the reference CSV file.
        ref_file (str): Name of the CSV file.
        input_interval (list[datetime.datetime]): ``[date_i, date_f]``
            simulation interval to cover.
        target_dir (str): Directory where the CSV file is linked.
        tracer: Unused directly, kept for interface consistency with other
            flux plugins.
        component: Unused, kept for interface consistency with other fetch
            functions.

    Returns:
        tuple[dict, dict]: ``(list_files, list_dates)``, each keyed by the
        start of a day within `input_interval`, mapping to the (repeated)
        CSV file path and the list of hourly ``[start, end]`` date-interval
        pairs within that day.
    """

    ref_file = f"{ref_dir}/{ref_file}"

    # Read the CSV file
    ds = pd.read_csv(ref_file, sep=";", parse_dates=["datei", "datef"])

    list_period_dates = pd.date_range(input_interval[0], input_interval[1], freq="1D")
    list_dates = {}
    list_files = {}
    for dd in list_period_dates:
        list_hours = pd.date_range(dd, dd + datetime.timedelta(hours=23), freq="1h")
        list_dates[dd] = [[hh, hh + datetime.timedelta(hours=1)] for hh in list_hours]

        # Generate list files and list_dates
        list_files[dd] = len(list_hours) * [ref_file]
        # list_dates = {input_interval[0]: [
        #     [d0, d1] for d0, d1 in zip(ds["datei"], ds["datef"])
        # ]}

    # Link to workdir
    target_file = f"{target_dir}/{os.path.basename(ref_file)}"
    path.link(ref_file, target_file)

    return list_files, list_dates