Prolog lazy streams using delimited continuations

delstreams.pl

:- module(delstreams).

ask(_, 0, _) :- throw(asked_closed_stream).
ask(El, Stream, Stream2) :- reset(Stream, El, Stream2), (Stream2 \= 0).

take(0, _) :- !.
take(N, S) :- N > 0, ask(El, S, S1), shift(El), succ(N1, N), take(N1, S1).

drop(0, S) :- !, call(S).
drop(N, S) :- N > 0, ask(_, S, S1), succ(N1, N), drop(N1, S1).

const(Element) :- shift(Element), const(Element).

repeat(N, Element) :- N > 0,
  shift(Element),
  succ(N0, N),
  repeat(N0, Element).
repeat(_, _) :- !, fail.

nat(From) :- shift(From), succ(From, From1), nat(From1).

list([]) :- !.
list([X | Xs]) :- shift(X), list(Xs).

to_list(S, [El | Rest]) :- ask(El, S, S1), !, to_list(S1, Rest).
to_list(_, []).

map(Goal, S) :-
  ask(X, S, E),
  call(Goal, X, Y),
  shift(Y),
  map(Goal, E).

map2(Predicate, S1, S2) :- 
  ask(X, S1, E1),
  ask(Y, S2, E2),
  call(Predicate, X, Y, Result),
  shift(Result),
  map2(Predicate, E1, E2).

concat(S1, S2) :- 
  (ask(X, S1, E) -> shift(X), concat(E, S2) ; call(S2)).

% Given input stream S, makes a new stream with moving average of the sliding window of size N
moving_average(N, S) :-
  moving_average_(N, 0, S, concat(repeat(N, -), S)).

moving_average_(N, Sum, Ins, Outs) :-
  ask(I, Ins, Ins1),
  ask(O, Outs, Outs1), !,
  (O = - -> Sum1 is Sum + I ; (Sum1 is Sum + I - O, Avg is rdiv(Sum,N), shift(Avg))),
  moving_average_(N, Sum1, Ins1, Outs1).

moving_average_(N, Sum, _, _) :-
  % Input stream has reached the end, emit last average and then close the MA stream.
  Avg is rdiv(Sum, N), shift(Avg), fail.

% Examples
% :- N = nat(0), N3 = drop(3, N), to_list(take(3, map2(plus, N, N3)), L).
% % L = [3, 5, 7].
% :- MA = moving_average(3, list([10,10,10,20,20,20,30,40,50,60,70])), to_list(MA, L)
% % L = [10, 40 rdiv 3, 50 rdiv 3, 20, 70 rdiv 3, 30, 40, 50, 60].