ldcast/analysis/histogram.py

import concurrent.futures
import multiprocessing
import os

import netCDF4
import numpy as np
from scipy.interpolate import interp1d

from ..features.io import load_batch, decode_saved_var_to_rainrate


def histogram(observation, forecasts, bins):
    N_bins = len(bins)-1
    N_timesteps = observation.shape[2]
    obs_hist = np.zeros((N_bins, N_timesteps), dtype=np.uint64)
    fc_hist = np.zeros((N_bins, N_timesteps), dtype=np.uint64)
    
    for t in range(observation.shape[2]):
        obs = observation[:,:,t,...].flatten()
        fc = forecasts[:,:,t,...].flatten()
        obs_hist[:,t] = np.histogram(obs, bins=bins)[0]
        fc_hist[:,t] = np.histogram(fc, bins=bins)[0]

    return (obs_hist, fc_hist)


def hist_from_file(fn, bins):
    print(fn)
    (_, y, y_pred) = load_batch(fn, threshold=bins[0])
    return histogram(y, y_pred, bins)


def save_histogram_for_dataset(data_dir, result_fn, bins=(0.05,120,100)):
    files = sorted(os.listdir(data_dir))
    files = [os.path.join(data_dir,fn) for fn in files]

    bins = np.exp(np.linspace(np.log(bins[0]), np.log(bins[1]), bins[2]))
    bins = np.hstack((0, bins))

    N_threads = multiprocessing.cpu_count()
    with concurrent.futures.ProcessPoolExecutor(N_threads) as executor:
        futures = []
        for fn in files:
            args = (hist_from_file, fn, bins)
            futures.append(executor.submit(*args))

        (obs_hist, fc_hist) = zip(*(f.result() for f in futures))

    obs_hist = sum(obs_hist)
    fc_hist = sum(fc_hist)

    with netCDF4.Dataset(result_fn, 'w') as ds:
        ds.createDimension("dim_bin", obs_hist.shape[0])
        ds.createDimension("dim_time_future", obs_hist.shape[1])
        var_params = {"zlib": True, "complevel": 1}

        obs_var = ds.createVariable(
            f"obs_hist", np.uint64,
            ("dim_bin", "dim_time_future"),
            **var_params
        )
        obs_var[:] = obs_hist

        fc_var = ds.createVariable(
            f"fc_hist", np.uint64,
            ("dim_bin", "dim_time_future"),
            **var_params
        )
        fc_var[:] = fc_hist

        ds.createDimension("dim_bin_edge", len(bins))
        bin_var = ds.createVariable(
            f"bins", np.float64,
            ("dim_bin_edge",),
            **var_params
        )
        bin_var[:] = bins


def load_histogram(fn):
    with netCDF4.Dataset(fn, 'r') as ds:
        obs_hist = np.array(ds["obs_hist"][:], copy=False)
        fc_hist = np.array(ds["fc_hist"][:], copy=False)
        bins = np.array(ds["bins"][:], copy=False)

    return (obs_hist, fc_hist, bins)


class ProbabilityMatch:
    def __init__(self, obs_hist, fc_hist, bins):
        obs_c = obs_hist.cumsum()
        obs_c = obs_c / obs_c[-1]
        fc_c = fc_hist.cumsum()
        fc_c = fc_c / fc_c[-1]

        self.obs_cdf = interp1d(np.hstack((0,obs_c)), bins, fill_value='extrapolate')
        self.fc_cdf = interp1d(bins, np.hstack((0,fc_c)), fill_value='extrapolate')

    def __call__(self, x):
        return self.obs_cdf(self.fc_cdf(x))


def probability_match_timesteps(obs_hist, fc_hist, bins):
    num_timesteps = obs_hist.shape[1]
    return [
        ProbabilityMatch(obs_hist[:,t], fc_hist[:,t], bins)
        for t in range(num_timesteps)
    ]