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

import datetime
import glob
from logging import debug, info, warn
import os
import calendar
from pathlib import Path
import numpy as np
from logging import info




def find_valid_file(ref_dir, file_format, dd, ref_dir_next, ref_dir_previous=False):
    # 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("Did not find any valid files in {} "
                 "with format {}"
                 .format(ref_dir_previous, 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("Did not find any valid files in {} "
                 "with format {}"
                 .format(ref_dir_previous, 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 len(list_files) == 0:
        raise Exception("Did not find any valid files in '{}' "
                        "with format '{}'. Please check your yml file"
                        .format(ref_dir, file_format))
    
    # Compute deltas
    mask = (list_dates - ref_date) <= datetime.timedelta(0)

    if not np.any(mask):
        warn("Did not find any date in '{}' starting before {} "
            "for the available files found: '{}'."
            .format(list_dates, ref_date, list_files)
        )
        warn("Assuming the first date {} is valid already since {}."
            .format(list_dates[0], ref_date))
        # Choose the last date and file
        file_ref1 = Path(ref_dir, list_files[0])
        date_ref1 = list_dates[0]
    else:
        # find nearest previous date
        file_ref1 = Path(ref_dir, list_files[mask][np.argmax(list_dates[mask])])
        date_ref1 = list_dates[mask].max()

    mask = (list_dates - ref_date) >= datetime.timedelta(0)
    if not np.any(mask):
        warn("Did not find any date in '{}' starting after {} "
            "for the available files found: '{}'."
            .format(list_dates, ref_date, list_files)
        )
        warn("Assuming the last date {} is valid until {}."
            .format(list_dates[-1], ref_date))
        # Choose the last date and file
        file_ref2 = Path(ref_dir, list_files[-1])
        date_ref2 = list_dates[-1]
    else:
        # find nearest next date
        file_ref2 = Path(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 [str(file_ref1), str(file_ref2)], [dd1, dd2]