Source code for pycif.plugins.datastreams.fluxes.VPRM1km_nc.utils

import datetime
import glob
import os
import calendar
import numpy as np
from logging import info
from .....utils.check.errclass import CifError




def find_valid_file(ref_dir, file_format, dd, ref_dir_next,
                    ref_dir_previous=False):
    """Find the nearest available files bracketing a given date.

    Lists candidate files in `ref_dir` matching `file_format`, extending
    the search to `ref_dir_next`/`ref_dir_previous` when `dd` is close to a
    month boundary (within 3 hours), sorts them by parsed date, and returns
    the file/date pair nearest to `dd` on-or-before and on-or-after it.

    Args:
        ref_dir (str): Directory to search for candidate files.
        file_format (str): ``strftime``-style filename pattern used both to
            match files and parse their embedded date.
        dd (datetime.datetime): Reference date to bracket.
        ref_dir_next (str): Directory for the following month, searched in
            addition to `ref_dir` when `dd` is near a month boundary.
        ref_dir_previous (str or bool): Directory for the previous month,
            searched in addition to `ref_dir` when `dd` is near a month
            boundary; ``False`` to skip.

    Returns:
        tuple[list[str], list[datetime.datetime]]: ``([file1, file2], [dd1, dd2])``,
        the paths and adjusted dates of the nearest file on-or-before and
        on-or-after `dd`.

    Raises:
        CifError: If no candidate files are found at all.
    """
    # Get all files and dates matching the file and format
    list_files_orig = os.listdir(ref_dir)

    # Convert ref date
    ref_date = datetime.datetime.strptime(dd.strftime(file_format), file_format)
    previous_date = ref_date - datetime.timedelta(hours=3)
    if previous_date.month < ref_date.month and ref_dir_previous:
        try:
            list_files_orig += os.listdir(ref_dir_previous)
        except:
            info(f"Did not find any valid files in {ref_dir_previous} with format {file_format}")

    next_date = ref_date + datetime.timedelta(hours=3)
    if next_date.month > ref_date.month:
        try:
            list_files_orig += os.listdir(ref_dir_next)
        except:
            info(f"Did not find any valid files in {ref_dir_previous} with format {file_format}")

    list_dates_cur = []
    list_files_cur = []
    for f in list_files_orig:
        try:
            if f.find('idx') < 0:
                list_dates_cur.append(
                    datetime.datetime.strptime(f, file_format))
                list_files_cur.append(f)
        except:
            continue

    list_files = np.array(list_files_cur)
    list_dates = np.array(list_dates_cur)

    # Sorting along dates
    isort = np.argsort(list_dates)
    list_dates = list_dates[isort]
    list_files = list_files[isort]

    if list_files == []:
        raise CifError(f"Did not find any valid files in {ref_dir} with format {file_format}. Please check your yml file")

    # Compute deltas
    mask = (list_dates - ref_date) <= datetime.timedelta(0)

    # find nearest previous date
    file_ref1 = ref_dir + list_files[mask][np.argmax(list_dates[mask])]
    date_ref1 = list_dates[mask].max()

    mask = (list_dates - ref_date) >= datetime.timedelta(0)
    # find nearest next date
    file_ref2 = ref_dir + list_files[mask][np.argmin(list_dates[mask])]
    date_ref2 = list_dates[mask].min()

    # Reconvert to original date
    dd1 = dd + (date_ref1 - ref_date)
    dd2 = dd + (date_ref2 - ref_date)

    return [file_ref1, file_ref2], [dd1, dd2]