from typing import Any, List, Optional

from reactivex import Observable, abc
from reactivex.scheduler import NewThreadScheduler
from reactivex.typing import RelativeTime

new_thread_scheduler = NewThreadScheduler()


def to_marbles(
    timespan: RelativeTime = 0.1, scheduler: Optional[abc.SchedulerBase] = None
):
    def to_marbles(source: Observable[Any]) -> Observable[str]:
        """Convert an observable sequence into a marble diagram string.

        Args:
            timespan: [Optional] duration of each character in second.
                If not specified, defaults to 0.1s.
            scheduler: [Optional] The scheduler used to run the the input
                sequence on.

        Returns:
            Observable stream.
        """

        def subscribe(
            observer: abc.ObserverBase[str],
            scheduler: Optional[abc.SchedulerBase] = None,
        ):
            scheduler = scheduler or new_thread_scheduler

            result: List[str] = []
            last = scheduler.now

            def add_timespan():
                nonlocal last

                now = scheduler.now
                diff = now - last
                last = now
                secs = scheduler.to_seconds(diff)
                timespan_ = scheduler.to_seconds(timespan)
                dashes = "-" * int((secs + timespan_ / 2.0) * (1.0 / timespan_))
                result.append(dashes)

            def on_next(value: Any) -> None:
                add_timespan()
                result.append(stringify(value))

            def on_error(exception: Exception) -> None:
                add_timespan()
                result.append(stringify(exception))
                observer.on_next("".join(n for n in result))
                observer.on_completed()

            def on_completed():
                add_timespan()
                result.append("|")
                observer.on_next("".join(n for n in result))
                observer.on_completed()

            return source.subscribe(on_next, on_error, on_completed)

        return Observable(subscribe)

    return to_marbles


def stringify(value: Any) -> str:
    """Utility for stringifying an event."""
    string = str(value)
    if len(string) > 1:
        string = "(%s)" % string

    return string


__all__ = ["stringify"]