Source code for pycif.plugins.datastreams.meteos.lmdz_massflx_old.fetch

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from logging import debug, info
from pathlib import Path

import numpy as np
import pandas as pd

from .....utils import path

_filetype_list = (
    ("fluxstoke", ""),
    ("fluxstokev", ""),
    ("phystoke", ""),
    ("phystoke", "_csr"),
)




def fetch(
    ref_dir,
    ref_file,
    input_interval,
    target_dir,
    tracer=None,
    **kwargs,
):
    """Reads meteorology and links to the working directory"""
    info(f"Linking meteo files from {ref_dir} to {target_dir}")

    datei, datef = input_interval

    ref_datei = pd.Timestamp(year=datei.year, month=datei.month, day=1)
    list_period_dates = pd.date_range(ref_datei, datef, freq=tracer.file_freq)
    list_period_dates = list_period_dates.to_pydatetime()

    if list_period_dates[0] < datei:
        list_period_dates[0] = datei

    if list_period_dates[-1] < datef:
        list_period_dates = np.append(list_period_dates, datef)

    # Link defstoke file for legacy LMDZ/acc plugin
    source = Path(getattr(tracer, "defstoke_file", "defstoke.nc"))
    if not source.is_absolute():
        source = Path(ref_dir, source)

    if source.is_file():
        target = Path(target_dir, source.name)
        path.link(source, target)

    list_dates = {}
    list_files = {}

    # Link meteo files
    for date in list_period_dates:
        ref_date = pd.Timestamp(year=date.year, month=date.month, day=1)

        for prefix, suffix in _filetype_list:
            if prefix == "defstoke":
                source = Path(getattr(tracer, "defstoke_file", "defstoke.nc"))
                if not source.is_absolute():
                    source = Path(ref_dir, source)
            else:
                filename = f"{prefix}.{date:an%Y.m%m}{suffix}.nc"
                source = Path(ref_dir, filename)

            if not source.is_file():
                continue

            # Link file
            target = Path(target_dir, source.name)
            path.link(source, target)

            # Get dates in the file
            if date not in list_dates:
                # TODO: need to get that from 'nday_step' in LMDZ model plugin
                nday_step = 8
                n_time = nday_step * ref_date.days_in_month

                list_dates[date] = [
                    [d, d + pd.Timedelta(hours=3)]
                    for d in pd.date_range(ref_date, periods=n_time, freq="3h")
                ]
                list_files[date] = len(list_dates[date]) * [source.name]

    for key, val in list_dates.items():
        debug(f"{key}:")
        for di, df in val:
            debug(f"  {di} - {df}")

    return list_files, list_dates