Source code for pycif.plugins.controlvects.standard.utils.build_lsm

import numpy as np
import datetime
import itertools

from .....utils.netcdf import readnc
from ....transforms.system.fromcontrol.utils.scalemaps import map2scale
from .....utils.path import init_dir
from .....utils.check.errclass import CifError, CifValueError




def build_lsm(corr, hresol, tracer):

    # First check proper definition of sigmas
    sigma_land = getattr(corr, "sigma_land", -1)
    sigma_sea = getattr(corr, "sigma_sea", -1)

    if sigma_land == sigma_sea == -1:
        raise CifError(
            "`sigma_land` and `sigma_sea` are both missing"
            "to compute horizontal correlations with a land-sea mask. \n"
            "One of the two can be missing if no correlations are to be accounted for "
            "in the corresponding land/sea area, but at least one must be specified"
        )

    # Different treatment between regions and other resolutions
    if hresol == "regions":
        if getattr(tracer, 'regions_lsm', False):
            landseamask = tracer.regions_lsmask

        landseamask = map2scale(
            landseamask[np.newaxis, np.newaxis, :, :],
            tracer, tracer.domain, region_scale_area=False,
            region_max_val=True).flatten()

        return landseamask

    # Now exclude for non hpixel resolution
    if hresol != "hpixels":
        raise CifValueError("Can't manage land sea mask for non hpixel or region resolutions. "
                         f"'{hresol}' was specified")

    # Carry on with processing pixel resolution
    if hasattr(corr, 'lsm_infos'):
        lsm_infos = corr.lsm_infos

        # Initialize target repository for linking
        target_dir = f"{tracer.workdir}/controlvect/landseamask/"
        init_dir(target_dir)

        list_files, list_dates = lsm_infos.fetch(
            lsm_infos.dir,
            lsm_infos.file,
            [datetime.datetime(1970, 1, 1),
             datetime.datetime(1970, 1, 1)],
            target_dir,
            tracer=lsm_infos,
        )

        all_files = list(itertools.chain(*list_files.values()))
        all_dates = np.array(list(itertools.chain(*list_dates.values())))

        # Cannot carry on if no files
        if all_files == []:
            raise CifValueError(
                'Cannot read land sea mask as no file fits description'
            )

        # Now read the variable
        landseamask = lsm_infos.read(
            'lsm',
            lsm_infos.varname,
            [all_dates[0]],
            [all_files[0]],
            tracer=lsm_infos,
        ).values[0, 0]

    else:
        file_lsm = getattr(corr, "filelsm")
        landseamask = readnc(file_lsm, ["lsm"]).flatten().data

    return landseamask, sigma_land, sigma_sea