dvcompute 2.0.0

// Copyright (c) 2020-2022  David Sorokin <davsor@mail.ru>, based in Yoshkar-Ola, Russia
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

use std::rc::Rc;

use crate::simulation::parameter::*;
use crate::simulation::parameter::random::*;
use crate::simulation::observable::*;
use crate::simulation::observable::disposable::*;
use crate::simulation::observable::source::*;
use crate::simulation::simulation::*;
use crate::simulation::composite::*;
use crate::simulation::process::*;

use dvcompute_utils::simulation::arrival::*;
use dvcompute_utils::grc::Grc;

/// Return an observable of random events that arrive with the specified delay.
pub fn new_random_observable<T, F, M>(f: F) -> impl Composite<Item = ObservableBox<Arrival<T>>>
    where F: Fn() -> M + 'static,
          M: Parameter<Item = (f64, T)> + 'static,
          T: 'static
{
    let source = Grc::new(ObservableSource::new());
    ProcessId::new()
        .into_composite()
        .and_then(move |pid| {
            let pid = Grc::new(pid);
            loop_random_observable(f, source.clone(), None)
                .run_using_id(pid.clone())
                .into_composite()
                .and_then(move |()| {
                    disposable_composite({
                        cons_disposable(move |p| {
                            ProcessId::cancel(pid)
                                .call_event(p)
                        })
                    })
                })
                .and_then(move |()| {
                    return_composite(source.publish().into_boxed())
                })
        })
}

/// The computation loop for `new_random_observable`.
fn loop_random_observable<T, F, M>(f: F, source: Grc<ObservableSource<Arrival<T>>>, t0: Option<f64>) -> ProcessBox<()>
    where F: Fn() -> M + 'static,
          M: Parameter<Item = (f64, T)> + 'static,
          T: 'static
{
    let p = f();
    p.into_process()
        .and_then(move |(delay, a)| {
            let delay = delay.max(0.0);
            hold_process(delay)
                .and_then(move |()| {
                    time_event()
                        .into_process()
                        .and_then(move |t2| {
                            let arrival = Arrival {
                                value: a,
                                time: t2,
                                delay: {
                                    match t0 {
                                        None => None,
                                        Some(_) => Some(delay)
                                    }
                                }
                            };
                            source.trigger(arrival)
                                .into_process()
                                .and_then(move |()| {
                                    loop_random_observable(f, source, Some(t2))
                                })
                        })
                })
        })
        .into_boxed()
}

/// An observable with random time delays distributed uniformly.
#[inline]
pub fn new_random_uniform_observable(min: f64, max: f64) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_uniform_parameter(min, max)
            .map(|x| (x, x))
    })
}

/// An observable with random integer time delays distributed uniformly.
#[inline]
pub fn new_random_int_uniform_observable(min: isize, max: isize) -> impl Composite<Item = ObservableBox<Arrival<isize>>> {
    new_random_observable(move || {
        random_int_uniform_parameter(min, max)
            .map(|x| (x as f64, x))
    })
}

/// An observable with random time delays from the triangular distribution.
#[inline]
pub fn new_random_triangular_observable(min: f64, median: f64, max: f64) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_triangular_parameter(min, median, max)
            .map(|x| (x, x))
    })
}

/// An observable with random time delays distributed normally.
#[inline]
pub fn new_random_normal_observable(mu: f64, nu: f64) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_normal_parameter(mu, nu)
            .map(|x| {
                let x = x.max(0.0);
                (x, x)
            })
    })
}

/// An observable with random time delays distributed lognormally.
#[inline]
pub fn new_random_log_normal_observable(mu: f64, nu: f64) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_log_normal_parameter(mu, nu)
            .map(|x| (x, x))
    })
}

/// An observable with exponential random time delays with the specified mean.
#[inline]
pub fn new_random_exponential_observable(mu: f64) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_exponential_parameter(mu)
            .map(|x| (x, x))
    })
}

/// An observable with random time delays from the Erlang distribution with the specified scale
/// (the reciprocal of the rate) and integer shape.
#[inline]
pub fn new_random_erlang_observable(scale: f64, shape: isize) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_erlang_parameter(scale, shape)
            .map(|x| (x, x))
    })
}

/// An observable with the Poisson random time delays with the specified mean.
#[inline]
pub fn new_random_poisson_observable(mu: f64) -> impl Composite<Item = ObservableBox<Arrival<isize>>> {
    new_random_observable(move || {
        random_poisson_parameter(mu)
            .map(|x| (x as f64, x))
    })
}

/// An observable with binomial random time delays with the specified probability and trials.
#[inline]
pub fn new_random_binomial_observable(prob: f64, trials: isize) -> impl Composite<Item = ObservableBox<Arrival<isize>>> {
    new_random_observable(move || {
        random_binomial_parameter(prob, trials)
            .map(|x| (x as f64, x))
    })
}

/// An observable with random time delays from the Gamma distribution.
#[inline]
pub fn new_random_gamma_observable(kappa: f64, theta: f64) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_gamma_parameter(kappa, theta)
            .map(|x| (x, x))
    })
}

/// An observable with random time delays from the Beta distribution.
#[inline]
pub fn new_random_beta_observable(alpha: f64, beta: f64) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_beta_parameter(alpha, beta)
            .map(|x| (x, x))
    })
}

/// An observable with random time delays from the Weibull distribution.
#[inline]
pub fn new_random_weibull_observable(alpha: f64, beta: f64) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_weibull_parameter(alpha, beta)
            .map(|x| (x, x))
    })
}

/// An observable with random time delays from the specified discrete distribution.
#[inline]
pub fn new_random_discrete_observable(dpdf: Rc<Vec<(f64, f64)>>) -> impl Composite<Item = ObservableBox<Arrival<f64>>> {
    new_random_observable(move || {
        random_discrete_parameter(dpdf.clone())
            .map(|x| (x, x))
    })
}