dvcompute_branch/simulation/observable/

mod.rs

// 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::marker::PhantomData;

use crate::simulation;
use crate::simulation::point::Point;
use crate::simulation::event::*;
use crate::simulation::ref_comp::RefComp;

use dvcompute_utils::grc::Grc;

/// Defines the disposable computation.
pub mod disposable;

/// Defines the observer computation.
pub mod observer;

/// Defines the observable source.
pub mod source;

/// Random observable computations.
pub mod random;

/// Defines the signal history.
pub mod history;

/// Additional operations.
pub mod ops;

use self::disposable::*;
use self::observer::*;
use self::source::*;

// /// Construct a new `Observable` computation by the specified function.
// #[inline]
// pub fn cons_observable<F, T>(f: F) -> Cons<F, T>
//     where F: FnOnce(ObserverBox<T, ()>) -> EventBox<DisposableBox>
// {
//      Cons { f: f, phantom: PhantomData }
// }

/// Return an empty observable that emits no messages.
#[inline]
pub fn empty_observable<T>() -> Empty<T> {
    Empty { phantom: PhantomData }
}

/// Delay the `Observable` computation.
#[inline]
pub fn delay_observable<F, M>(f: F) -> Delay<F, M>
    where F: FnOnce() -> M + 'static,
          M: Observable + 'static
{
    Delay { f: f, _phantom: PhantomData }
}

/// Return a source of signal that is triggered in the integration time points.
/// It should be called in the start time only.
#[inline]
pub fn new_observable_source_in_integ_times() -> NewSourceInIntegTimes {
    NewSourceInIntegTimes {}
}

/// Return a signal that is triggered in the start time point.
/// It should be called in the start time only.
#[inline]
pub fn new_observable_in_start_time() -> NewInStartTime {
    NewInStartTime {}
}

/// Return a source of signal that is triggered in the start time point.
/// It should be called in the start time only.
#[inline]
pub fn new_observable_source_in_start_time() -> NewSourceInStartTime {
    NewSourceInStartTime {}
}

/// Return a signal that is triggered in the stop time point.
/// It should be called in the start time only.
#[inline]
pub fn new_observable_in_stop_time() -> NewInStopTime {
    NewInStopTime {}
}

/// Return a source of signal that is triggered in the stop time point.
/// It should be called in the start time only.
#[inline]
pub fn new_observable_source_in_stop_time() -> NewSourceInStopTime {
    NewSourceInStopTime {}
}

/// Trace the computation.
#[inline]
pub fn trace_observable<M>(msg: String, comp: M) -> Trace<M>
    where M: Observable
{
    Trace { comp: comp, msg: msg}
}

/// The computation that allows notifying about the events.
pub trait Observable {

    /// The type of messages to notify in the current modeling time.
    type Message;

    /// Subscribe the observer.
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static;

    /// Map the current computation using the specified transform.
    #[inline]
    fn map<B, F>(self, f: F) -> Map<Self, B, F>
        where Self: Sized,
              F: Fn(&Self::Message) -> B + Clone + 'static,
              Self::Message: Clone + 'static,
              B: 'static
    {
        Map { comp: self, f: f, phantom: PhantomData }
    }

    /// Map the current computation using the specified transform computation.
    #[inline]
    fn mapc<U, B, F>(self, f: F) -> MapC<Self, U, B, F>
        where Self: Sized,
              F: Fn(&Self::Message) -> U + Clone + 'static,
              U: Event<Item = B>,
              Self::Message: Clone + 'static,
              B: 'static
    {
        MapC { comp: self, f: f, _phantom1: PhantomData, _phantom2: PhantomData }
    }

    /// Filter the messages.
    #[inline]
    fn filter<F>(self, f: F) -> Filter<Self, F>
        where Self: Sized,
              F: Fn(&Self::Message) -> bool + Clone + 'static,
              Self::Message: Clone + 'static
    {
        Filter { comp: self, f: f }
    }

    /// Filter the messages within computation.
    #[inline]
    fn filterc<U, F>(self, f: F) -> FilterC<Self, U, F>
        where Self: Sized,
              F: Fn(&Self::Message) -> U + Clone + 'static,
              U: Event<Item = bool>,
              Self::Message: Clone + 'static
    {
        FilterC { comp: self, f: f, _phantom: PhantomData }
    }

    /// Hold the receiving of messages for the specified time interval.
    #[inline]
    fn hold(self, dt: f64) -> Hold<Self>
        where Self: Sized,
              Self::Message: Clone + 'static
    {
        Hold { comp: self, dt: dt }
    }

    /// Hold the receiving of messages for intervals calculated within `Event` computation.
    #[inline]
    fn holdc<U, F>(self, f: F) -> HoldC<Self, U, F>
        where Self: Sized,
              Self::Message: Clone + 'static,
              F: Fn() -> U + Clone + 'static,
              U: Event<Item = f64>
    {
        HoldC { comp: self, f: f, _phantom: PhantomData }
    }

    /// Merge the current computation with another one within the resulting computation.
    #[inline]
    fn merge<U>(self, other: U) -> Merge<Self, U>
        where Self: Sized,
              Self::Message: Clone + 'static,
              U: Observable<Message = Self::Message>
    {
        Merge { comp: self, other: other }
    }

    /// Convert into a boxed value.
    #[inline]
    fn into_boxed(self) -> ObservableBox<Self::Message>
        where Self: Sized + Clone + 'static
    {
        ObservableBox::new(move |observer: ObserverBox<Self::Message, ()>| {
            self.subscribe(observer.into_observer())
        })
    }
}

/// Allows converting to `Observable` computations.
pub trait IntoObservable {

    /// The target computation.
    type Observable: Observable<Message = Self::Message>;

    /// The type of messages about which the computation notifies.
    type Message;

    /// Convert to the `Observable` computation.
    fn into_observable(self) -> Self::Observable;
}

impl<M: Observable> IntoObservable for M {

    type Observable = M;

    type Message = M::Message;

    #[inline]
    fn into_observable(self) -> Self::Observable {
        self
    }
}

/// It represents the boxed `Observable` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
pub struct ObservableBox<T> {
    f: Box<dyn ObservableFnBoxClone<T>>
}

impl<T> ObservableBox<T> {

    /// Create a new boxed computation.
    #[doc(hidden)]
    #[inline]
    fn new<F>(f: F) -> Self
        where F: FnOnce(ObserverBox<T, ()>) -> EventBox<DisposableBox> + Clone + 'static,
              T: 'static
    {
        ObservableBox {
            f: Box::new(f)
        }
    }

    /// Call the boxed function.
    #[doc(hidden)]
    #[inline]
    pub fn call_box(self, arg: (ObserverBox<T, ()>,)) -> EventBox<DisposableBox> {
        let ObservableBox { f } = self;
        f.call_box(arg)
    }
}

impl<T> Clone for ObservableBox<T> {

    #[inline]
    fn clone(&self) -> Self {
        ObservableBox {
            f: self.f.call_clone()
        }
    }
}

impl<T> Observable for ObservableBox<T> where T: 'static {

    type Message = T;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        self.call_box((observer.into_boxed(),))
    }

    #[inline]
    fn into_boxed(self) -> ObservableBox<Self::Message>
        where Self: Sized + Clone + 'static
    {
        self
    }
}

/// A trait to support the stable version of Rust, where there is no `FnBox`.
trait ObservableFnBox<T> {

    /// Call the corresponding function.
    fn call_box(self: Box<Self>, args: (ObserverBox<T, ()>,)) -> EventBox<DisposableBox>;
}

impl<T, F> ObservableFnBox<T> for F
    where F: FnOnce(ObserverBox<T, ()>) -> EventBox<DisposableBox>,
          T: 'static
{
    fn call_box(self: Box<Self>, args: (ObserverBox<T, ()>,)) -> EventBox<DisposableBox> {
        let this: Self = *self;
        this(args.0)
    }
}

/// A trait to implement a cloneable `FnBox`.
trait ObservableFnBoxClone<T>: ObservableFnBox<T> {

    /// Clone the function.
    fn call_clone(&self) -> Box<dyn ObservableFnBoxClone<T>>;
}

impl<T, F> ObservableFnBoxClone<T> for F
    where F: FnOnce(ObserverBox<T, ()>) -> EventBox<DisposableBox> + Clone + 'static,
          T: 'static
{
    fn call_clone(&self) -> Box<dyn ObservableFnBoxClone<T>> {
        Box::new(self.clone())
    }
}

/// The functor for the `Observable` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Map<M, B, F> {

    /// The current computation.
    comp: M,

    /// The transform.
    f: F,

    /// To keep the type parameter.
    phantom: PhantomData<B>
}

impl<M, B, F> Observable for Map<M, B, F>
    where M: Observable,
          F: Fn(&M::Message) -> B + Clone + 'static,
          M::Message: Clone + 'static,
          B: 'static
{
    type Message = B;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let comp     = self.comp;
        let f        = self.f;
        let observer = cons_observer(move |m: &M::Message, p: &Point| {
            let m2 = f(m);
            observer.call_observer(&m2, p)
        });

        comp.subscribe(observer)
    }
}

/// The transform for the `Observable` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct MapC<M, U, B, F> {

    /// The current computation.
    comp: M,

    /// The transform.
    f: F,

    /// To keep the type parameter.
    _phantom1: PhantomData<U>,

    /// To keep the type parameter.
    _phantom2: PhantomData<B>
}

impl<M, U, B, F> Observable for MapC<M, U, B, F>
    where M: Observable,
          F: Fn(&M::Message) -> U + Clone + 'static,
          U: Event<Item = B>,
          M::Message: Clone + 'static,
          B: 'static
{
    type Message = B;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let comp     = self.comp;
        let f        = self.f;
        let observer = cons_observer(move |m: &M::Message, p: &Point| {
            match f(m).call_event(p) {
                Result::Err(e) => Result::Err(e),
                Result::Ok(m2) => observer.call_observer(&m2, p)
            }
        });

        comp.subscribe(observer)
    }
}

/// The filter for the `Observable` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Filter<M, F> {

    /// The current computation.
    comp: M,

    /// The transform.
    f: F
}

impl<M, F> Observable for Filter<M, F>
    where M: Observable,
          F: Fn(&M::Message) -> bool + Clone + 'static,
          M::Message: Clone + 'static
{
    type Message = M::Message;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let comp     = self.comp;
        let f        = self.f;
        let observer = cons_observer(move |m: &M::Message, p: &Point| {
            if f(m) {
                observer.call_observer(m, p)
            } else {
                Result::Ok(())
            }
        });

        comp.subscribe(observer)
    }
}

/// The monadic filter for the `Observable` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct FilterC<M, U, F> {

    /// The current computation.
    comp: M,

    /// The transform.
    f: F,

    /// To keep the type parameter.
    _phantom: PhantomData<U>
}

impl<M, U, F> Observable for FilterC<M, U, F>
    where M: Observable,
          F: Fn(&M::Message) -> U + Clone + 'static,
          U: Event<Item = bool>,
          M::Message: Clone + 'static
{
    type Message = M::Message;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let comp     = self.comp;
        let f        = self.f;
        let observer = cons_observer(move |m: &M::Message, p: &Point| {
            match f(m).call_event(p) {
                Result::Err(e) => Result::Err(e),
                Result::Ok(true) => observer.call_observer(m, p),
                Result::Ok(false) => Result::Ok(())
            }
        });

        comp.subscribe(observer)
    }
}

/// Hold the receiving of messages within the `Observable` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Hold<M> {

    /// The current computation.
    comp: M,

    /// The delay interval.
    dt: f64
}

impl<M> Observable for Hold<M>
    where M: Observable,
          M::Message: Clone + 'static
{
    type Message = M::Message;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let comp = self.comp;
        let dt   = self.dt;
        let r    = Grc::new(RefComp::new(false));
        let observer = {
            let r = r.clone();
            let observer = Grc::new(observer);
            cons_observer(move |m: &M::Message, p: &Point| {
                let m = m.clone();
                let r = r.clone();
                let observer = observer.clone();
                let comp = cons_event(move |p| {
                    let x = r.read_at(p);
                    if !x {
                        observer.call_observer(&m, p)
                    } else {
                        Result::Ok(())
                    }
                });
                enqueue_event(p.time + dt, comp.into_boxed())
                    .call_event(p)
            })
        };
        let h = comp.subscribe(observer);
        let h = cons_disposable(move |p| {
            r.write_at(true, p);
            let h = h.call_event(p)?;
            h.dispose(p)
        }).into_boxed();
        cons_event(move |_p| { Result::Ok(h) })
            .into_boxed()
    }
}

/// Hold the receiving of messages within the `Observable` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct HoldC<M, U, F> {

    /// The current computation.
    comp: M,

    /// Get the delay interval within computation.
    f: F,

    /// The phantom parameter.
    _phantom: PhantomData<U>
}

impl<M, U, F> Observable for HoldC<M, U, F>
    where M: Observable,
          M::Message: Clone + 'static,
          F: Fn() -> U + Clone + 'static,
          U: Event<Item = f64>
{
    type Message = M::Message;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let comp = self.comp;
        let f    = self.f;
        let r    = Grc::new(RefComp::new(false));
        let observer = {
            let r = r.clone();
            let observer = Grc::new(observer);
            cons_observer(move |m: &M::Message, p: &Point| {
                let m = m.clone();
                let r = r.clone();
                let observer = observer.clone();
                let comp = cons_event(move |p| {
                    let x = r.read_at(p);
                    if !x {
                        observer.call_observer(&m, p)
                    } else {
                        Result::Ok(())
                    }
                });
                let dt = f();
                let dt = dt.call_event(p)?;
                enqueue_event(p.time + dt, comp.into_boxed())
                    .call_event(p)
            })
        };
        let h = comp.subscribe(observer);
        let h = cons_disposable(move |p| {
            r.write_at(true, p);
            let h = h.call_event(p)?;
            h.dispose(p)
        }).into_boxed();
        cons_event(move |_p| { Result::Ok(h) })
            .into_boxed()
    }
}

/// The merge of two `Observable` computations.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Merge<M, U> {

    /// The current computation.
    comp: M,

    /// Another computation.
    other: U,
}

impl<M, U> Observable for Merge<M, U>
    where M: Observable,
          M::Message: Clone + 'static,
          U: Observable<Message = M::Message>
{
    type Message = M::Message;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let observer  = Grc::new(observer);
        let comp      = self.comp;
        let other     = self.other;
        let observer1 = {
            let observer = observer.clone();
            cons_observer(move |m: &M::Message, p: &Point| {
                observer.call_observer(m, p)
            })
        };
        let observer2 = {
            cons_observer(move |m: &M::Message, p: &Point| {
                observer.call_observer(m, p)
            })
        };
        let disposable1 = comp.subscribe(observer1);
        let disposable2 = other.subscribe(observer2);

        disposable1.into_event().and_then(move |disposable1| {
            disposable2.into_event().map(move |disposable2| {
                let disposable1 = disposable1.into_disposable();
                let disposable2 = disposable2.into_disposable();

                disposable1.merge(disposable2).into_boxed()
            })
        }).into_boxed()
    }
}

/// The empty `Observable` computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Empty<T> {

    /// The transform.
    phantom: PhantomData<T>
}

impl<T> Observable for Empty<T> {

    type Message = T;

    #[inline]
    fn subscribe<O>(self, _observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let disposable = empty_disposable().into_boxed();
        return_event(disposable).into_boxed()
    }
}

/// Allows delaying the `Observable` computation by the specified function.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Delay<F, M> {

    /// Return the computation.
    f: F,

    /// To keep the type parameter.
    _phantom: PhantomData<M>
}

impl<F, M> Observable for Delay<F, M>
    where F: FnOnce() -> M + 'static,
          M: Observable + 'static
{
    type Message = M::Message;

    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let Delay { f, _phantom } = self;
        f().subscribe(observer)
    }
}

/// A source of signal in the integration time points.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct NewSourceInIntegTimes {}

impl Event for NewSourceInIntegTimes {

    type Item = Grc<ObservableSource<f64>>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let source  = Grc::new(source::ObservableSource::new());
        let source2 = source.clone();
        enqueue_events_with_integ_times(move || {
            let source = source.clone();
            cons_event(move |p| {
                source.trigger_at(&p.time, p)
            }).into_boxed()
        }).call_event(p)?;
        Result::Ok(source2)
    }
}

/// A source of signal in the start time point.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct NewSourceInStartTime {}

impl Event for NewSourceInStartTime {

    type Item = Grc<ObservableSource<f64>>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let source  = Grc::new(source::ObservableSource::new());
        let source2 = source.clone();
        enqueue_event(p.run.specs.start_time, {
            cons_event(move |p| {
                source.trigger_at(&p.time, p)
            }).into_boxed()
        }).call_event(p)?;
        Result::Ok(source2)
    }
}

/// A source of signal in the stop time point.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct NewSourceInStopTime {}

impl Event for NewSourceInStopTime {

    type Item = Grc<ObservableSource<f64>>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let source  = Grc::new(source::ObservableSource::new());
        let source2 = source.clone();
        enqueue_event(p.run.specs.stop_time, {
            cons_event(move |p| {
                source.trigger_at(&p.time, p)
            }).into_boxed()
        }).call_event(p)?;
        Result::Ok(source2)
    }
}

/// Trigger a signal in the integration time points.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct NewInIntegTimes {}

impl Event for NewInIntegTimes {

    type Item = ObservableBox<f64>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let comp   = NewSourceInIntegTimes {};
        let source = comp.call_event(p)?;
        Result::Ok(source.publish().into_boxed())
    }
}

/// Trigger a signal in the start time point.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct NewInStartTime {}

impl Event for NewInStartTime {

    type Item = ObservableBox<f64>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let comp   = NewSourceInStartTime {};
        let source = comp.call_event(p)?;
        Result::Ok(source.publish().into_boxed())
    }
}

/// Trigger a signal in the stop time point.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct NewInStopTime {}

impl Event for NewInStopTime {

    type Item = ObservableBox<f64>;

    #[doc(hidden)]
    #[inline]
    fn call_event(self, p: &Point) -> simulation::Result<Self::Item> {
        let comp   = NewSourceInStopTime {};
        let source = comp.call_event(p)?;
        Result::Ok(source.publish().into_boxed())
    }
}

/// Trace the computation.
#[must_use = "computations are lazy and do nothing unless to be run"]
#[derive(Clone)]
pub struct Trace<M> {

    /// The computation.
    comp: M,

    /// The message to print.
    msg: String
}

impl<M> Observable for Trace<M>
    where M: Observable
{
    type Message = M::Message;

    #[doc(hidden)]
    #[inline]
    fn subscribe<O>(self, observer: O) -> EventBox<DisposableBox>
        where O: Observer<Message = Self::Message, Item = ()> + Clone + 'static
    {
        let Trace { comp, msg } = self;
        comp.subscribe(trace_observer(msg, observer))
    }
}