view cpack/dml/api/perspectives.pl @ 0:718306e29690 tip

commiting public release

/* Part of DML (Digital Music Laboratory)
	Copyright 2014-2015 Samer Abdallah, University of London
	 
	This program is free software; you can redistribute it and/or
	modify it under the terms of the GNU General Public License
	as published by the Free Software Foundation; either version 2
	of the License, or (at your option) any later version.

	This program is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
	GNU General Public License for more details.

	You should have received a copy of the GNU General Public
	License along with this library; if not, write to the Free Software
	Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/

:- module(perspectives, []).

/** <module> VIS API Perspectives

   Todo

   - Chord sequences
   - Standardise data structures
*/
:- use_module(library(http/http_dispatch), [http_link_to_id/3]).
:- use_module(library(semweb/rdf_db)).
:- use_module(library(semweb/rdf_label)).
:- use_module(library(dcg_core)).
:- use_module(library(insist)).
:- use_module(library(computations)).
:- use_module(library(backend_json)).
:- use_module(library(dataset)).
:- use_module(library(memo)).
:- use_module(library(async)).
:- use_module(library(mlserver)).
:- use_module(api(dmlvis)).
:- use_module(api(archive)).

% :- setting(memoise_failures,boolean,false,"Whether or not to record failed computations to avoid retrying").
:- setting(default_recompute_policy,oneof([none,failed,force]),none,'Default policy on recomputing memoised computations').
:- setting(default_vamp_on_demand,boolean,false,'Default policy on doing VAMP computations on demand').

% registry of perspectives.
dmlvis:perspective( getRecordingPerspective,  Name, 
                    [+uri(URI),vamp_on_demand(V)-false|Params], 
                    cc(perspectives:rla(Pred,[vamp_on_demand(V)],URI))
                  ) :- rec_persp(Name, Params, Pred).

dmlvis:perspective( getCollectionPerspective, Name, 
                    [+cid(CID),recompute(R)-none,vamp_on_demand(V)-false,coverage(C)-summary|Params], 
                    cc(perspectives:cla(Pred,[recompute(R),vamp_on_demand(V),coverage(C)],CID))
                  ) :- coll_persp(Name, Params, Pred).

:- meta_predicate rla(2,+,+,-,-), cla(2,+,+,-,-).

rla(Pred,Opts,URI,Result,stable) :- 
   option(vamp_on_demand(V), Opts, false),
   with_global(vamp_on_demand, V, call(Pred,URI,Result)).

cla(Pred,Opts,CID,Result,stable) :- 
   check_collection(CID), 
   call(Pred,Opts,CID,Result1),
   option(coverage(Cov),Opts,full),
   insist(filter_coverage(Cov,Result1,Result), invalid_coverage_parameter(Cov)).

:- op(1050,xfy,=>).
G1 => G2 --> (call_dcg(G1) -> call_dcg(G2); []).

filter_coverage(full) --> [].
filter_coverage(summary) --> dtrans(coverage,C1,C2) => {summarise_coverage(C1,C2)}.
summarise_coverage --> foldl(replace_list_with_length,[failed,errors],[failed_count,errors_count]).
replace_list_with_length(Key) --> dtrans(Key,List,Length) => {length(List,Length)}.
replace_list_with_length(Key1,Key2) --> ddel(Key1,List) => {length(List,Len)}, dput(Key2,Len).
dtrans(Key,Val1,Val2,D1,D2) :- get_dict(Key,D1,Val1,D2,Val2).
ddel(Key,Val,D1,D2) :- del_dict(Key,D1,Val,D2).
dput(Key,Val,D1,D2) :- put_dict(Key,D1,Val,D2).
% dget(Key,Val,D,D) :- get_dict(Key,D,Val).

check_collection(CID) :-
   insist(dataset_size(CID,Size), unknown_collection(CID)),
   debug(dmlvis(perspective),'Doing collection level analysis on ~d items.',[Size]).

rec_persp( transcription,      [], output_link(transcription(0))).
rec_persp( transcription_fine, [], output_link(transcription(1))).
rec_persp( chords,             [], output_link(chords)).
rec_persp( chord_notes,        [], output_link(chord_notes)).
rec_persp( beatroot,           [], output_link(beats(beatroot))).
rec_persp( key,                [], output_link(key)).
rec_persp( key_tonic,          [], output_link(tonic)).
rec_persp( beats,              [], output_link(beats(qm))).
rec_persp( tempo,              [], output_link(tempo)).
rec_persp( chromagram,         [], output_link(chromagram)).
rec_persp( mfcc,               [], output_link(mfcc)).

rec_persp( spectrogram,          [offset(O)-0,length(L)-60], spectrogram_link(O,L)).
rec_persp( tempo_nonuniform,     [],                         nonuniform_tempo).
rec_persp( tempo_uniform,        [period(DT)-1,lang(L)-ml ], uniform_tempo(L,DT)).
rec_persp( tempo_normalised,     [num_samples(N)-20,lang(L)-ml ], normalised_tempo(L,N)).
rec_persp( chord_histogram,      [],                         chord_histogram).
rec_persp( midi_pitch_histogram, [weighting(W)-none],        pitch_histogram(W)).
rec_persp( pitch_histogram,      [weighting(W)-none, quant(Q)-5, min(Min)-0, max(Max)-127, lang(L)-ml ], 
           freq_histogram(L,Min,Max,Q,W)).
rec_persp( tempo_histogram,      [period(DT)-1, num_bins(N)-50, min(Min)-20, max(Max)-360, lang(L)-ml ], 
           tempo_histogram(L,DT,Min,Max,N)).

%% coll_persp(P:perspective(A), Params:list(param), Pred:pred(+options,+dataset,-A)) is nondet.
%
%  Database of collection perspectives. The first argument is an atom denoting a perspective
%  which returns results of type A. Params must be defined as in dmlvis:options_optspec/2.
%  Pred must accept a list of options and a collection (dataset) id and produce a result.
coll_persp( mean_tempo_curve,     [num_samples(N)-20,lang(L)-ml], mem(collection_tempo_curve(L,N))).
coll_persp( midi_pitch_histogram, [weighting(W)-none],            mem(collection_pitch_histogram(W))).
coll_persp( pitch_histogram,      [weighting(W)-none, quant(Q)-5, min(Min)-20, max(Max)-100, lang(L)-ml], 
            mem(collection_freq_histogram(L,Min,Max,Q,W))).
coll_persp( tempo_histogram,      [period(DT)-1, num_bins(N)-50, min(Min)-20, max(Max)-100, lang(L)-ml], 
            mem(collection_tempo_histogram(L,DT,Min,Max,N))).
coll_persp( pitch_lookup,         [+midi_pitch(P), weighting(W)-none, limit(Lim)-5000,offset(Off)-0], 
            nomem(collection_pitch_lookup(W,P,Lim,Off))).

% using python back-end
coll_persp( tonic_relative_pitch_class_histogram, [], 
            mem(py_hist(transcription_tonic_duration, tonic_norm_semitone_hist:aggregate, [opts{normalisation:piece}]))).
coll_persp( tonic_histogram,        [], mem(py_hist(tagged(tonic), key_tonic_hist:aggregate, []))).
coll_persp( pitch_class_histogram,  [], mem(py_hist(tagged(transcription), semitone_hist:aggregate, []))).
coll_persp( tuning_stats,           [], mem(py_cla(tagged(transcription(1)),tuning_stats:per_file,[]))).
coll_persp( tuning_stats_by_year,   [], mem(py_cla(transcription_date,tuning_stats_byyear:per_file,[]))).
coll_persp( places_hist,            [], nomem(py_cla(list_places,places_hist:per_file,[]))).
coll_persp( key_relative_chord_seq, 
            [  spm_minlen(MinLen)-2, spm_maxseqs(MaxSeqs)-500, spm_algorithm(Alg)-'CM-SPADE',
               spm_ignore_n(Ignn)-1, spm_maxtime(Smaxt)-60,    spm_minsupport(Smins)-50 ], 
            mem(py_cla( keys_chords,chord_seq_key_relative:aggregate, 
                        [opts{ spm_minlen:MinLen, spm_maxseqs:MaxSeqs, spm_algorithm:Alg,
                               spm_ignore_n:Ignn,spm_maxtime:Smaxt,spm_minsupport:Smins } ]))).

coll_persp( similarity,     
            [  sim_downsample(SimDown)-1,sim_clusters(SimClusters)-40,sim_reclimit(Limo)-2000,
               sim_type(SimType)-'euclidean',sim_features(SimFeat)-'chromagram',
               sim_compressor(SimComp)-'zlib'], 
            mem(py_cla(similarity_bundle,similarity:per_file,
                       [opts{sim_type:SimType,sim_clusters:SimClusters,sim_downsample:SimDown,
                             sim_reclimit:Limo,sim_features:SimFeat,sim_compressor:SimComp}]))).

% adaptor to ignore collection perspective options parameter
nomem(Goal,Opts,CID,Result) :- 
   option(vamp_on_demand(V), Opts, false),
   with_global(vamp_on_demand, V, 
      with_progress_stack(call(Goal,CID,Result))).
											   
dmlvis:param( recompute, [oneof([none,failed,force]), default(Def), 
                          description('Controls handling of memoised collection level results')]) :-
   setting(default_recompute_policy,Def).
dmlvis:param( vamp_on_demand, [boolean, default(Def), 
                           description('Whether to run VAMP plugins if results are not already available')]) :-
   setting(default_vamp_on_demand,Def).
dmlvis:param( coverage, [oneof([full,summary]), default(summary), 
                           description('How much detail to provide about recordings not successfully included in CLA')]).
dmlvis:param( offset,    [number, default(0), description('Offset into signal in seconds')]).
dmlvis:param( length,    [number, default(60), description('Length of signal extract in seconds')]).
dmlvis:param( weighting, [oneof([none,dur,vel]), default(none), description('Weighting for pitch_histogram perspective')]).
dmlvis:param( quant,     [nonneg, default(5), description('Subdivisions of a semitone for freq_histogram')]).
dmlvis:param( period,    [number, default(1), description('Sampling period in seconds')]).
dmlvis:param( num_bins,  [nonneg, default(50), description('Number of bins for histogram')]).
dmlvis:param( num_samples, [nonneg, default(50), description('Number of samples for normalised histogram')]).
dmlvis:param( max,       [nonneg, default(100), description('Max pitch for pitch histogram')]).
dmlvis:param( min,       [nonneg, default(20), description('Min pitch for pitch histogram')]).
dmlvis:param( lang,      [oneof([ml,r]), default(r), description('Numerical computations language')]).

/* chord sequence parameters */
dmlvis:param( spm_minlen,    [nonneg, default(2),          description('Minimum length of chord sequence')]).
dmlvis:param( spm_maxseqs,   [nonneg, default(500),        description('Maximum number of sequences to return')]).
dmlvis:param( spm_algorithm, [atom,   default('CM-SPADE'), description('CM-SPADE, TKS or ClaSP')]).
dmlvis:param( spm_ignore_n,   [nonneg, default(1),         description('Ignore failed chord detections')]).
dmlvis:param( spm_maxtime,   [nonneg, default(60),         description('Max. runtime for SPM algorithm')]).
dmlvis:param( spm_minsupport,   [nonneg, default(50),         description('Minimal Support in Percent')]).


/* similarity parameters */
dmlvis:param( sim_type, [atom,   default('euclidean'), description('Tpye of similarity measure: euclidean, compression')]).
dmlvis:param( sim_clusters, [nonneg,   default(40), description('Number of clusters for vector Quantisation (40-200)')]).
dmlvis:param( sim_downsample, [nonneg,   default(1), description('Downsample the audio analysis to a resolution of 1 second')]).
dmlvis:param( sim_reclimit, [nonneg,   default(2000), description('Maximum number of recordings in dataset')]).
dmlvis:param( sim_features, [atom,   default('chromagram'), description('Feature basis of the similarity estimation, any combination, separated by comma: chromagram,mfcc,chords')]).
dmlvis:param( sim_compressor, [atom,   default('zlib'), description('Compressor for similarity estimation: zlib, zxd')]).


:- rdf_meta transform_computation(+,r,r).
transform_computation(Class,In,Out) :- 
   (  transform(Class,Fn), computation(Fn,In,Out) *-> true
   ;  (  nb_current(vamp_on_demand,true)
      -> insist(transform(Class,Fn), unrecognised_transform_class(Class)), % picks first match 
         format(string(Desc),"Running computation ~w on ~w.",[Fn,In]),
         simple_task(Desc,computation_memo(Fn,In,Out))
      ;  throw(missing_computation(Class,In))
      )
   ).

:- rdf_meta transform_op(+,+,r,-).
transform_op(TName,Op,In,Out) :- 
   transform_computation(TName,In,X), 
   csv_op(Op,X,Out).

% ------- recording level perspectives ------------

spectrogram_link(Offs,Len,URI,_{image_url:Link}) :-
   http_link_to_id(spectrogram_window, [uri(URI), offset(Offs), length(Len)], Link).

output_link(TransformName,Input,_{csv:Output}) :-
   transform_computation(TransformName,Input,Output).

chord_histogram(URI,_{values:Chords, counts:Counts}) :-
   transform_op(chords,chord_hist,URI,Hist),
   unzip(Hist,Chords,Counts).

pitch_histogram(W,URI,_{values:NNs, counts:Counts}) :-
   transform_op(transcription,pitch_hist(W),URI,Hist),
   unzip(Hist,NNs,Counts).

freq_histogram(ml,Min,Max,Q,W,URI,_{edges:Edges, counts:Counts}) :-
   microtone_map(Min,Max,Q,Map),
   transform_op(transcription(1),freq_hist(Map,W),URI,Counts),
   map_edges(ml,Map,Edges).
freq_histogram(r,Min,Max,Q,W,URI,_{edges:Edges, counts:Counts}) :-
   microtone_map(Min,Max,Q,Map),
   transform_op(transcription(1),freq_hist_r(Map,W),URI,Counts),
   map_edges(r,Map,Edges).

nonuniform_tempo(URI,_{times:Times, values:Values}) :-
   transform_op(tempo,tempo,URI,Result),
   unzip(Result,Times,Values).

uniform_tempo(ml,DT,URI,_{times:Times, values:Values}) :-
   transform_op(tempo,uniform_tempo(DT),URI,Result),
   Result=Times-Values.

uniform_tempo(r,DT,URI,_{times:Times, values:Values}) :-
   transform_op(tempo,uniform_tempo_r(DT),URI,Result),
   Result=Times-Values.

normalised_tempo(ml,N,URI,_{times:Times, values:Values}) :-
   transform_op(tempo,normalised_tempo(N),URI,Result),
   Result=Times-Values.

normalised_tempo(r,N,URI,_{times:Times, values:Values}) :-
   transform_op(tempo,normalised_tempo_r(N),URI,Result),
   Result=Times-Values.

tempo_histogram(ml,DT,Min,Max,N,URI,_{edges:Edges, counts:Counts}) :-
   insist(Min>0, domain_error(min,"positive value",Min)), 
   Map=expmap(Min,Max,N),
   map_edges(ml,Map,Edges),
   transform_op(tempo,tempo_hist(DT,Map),URI,Result),
   Result=_-Counts.
tempo_histogram(r,DT,Min,Max,N,URI,_{edges:Edges, counts:Counts}) :-
   insist(Min>0, domain_error(min,"positive value",Min)), 
   Map=expmap(Min,Max,N),
   map_edges(r,Map,Edges),
   transform_op(tempo,tempo_hist_r(DT,Map),URI,Result),
   Result=_-Counts.


% ------- collection level perspectives ------------

collection_pitch_histogram(W,CID,Result) :-
   Min-Max = 20-100, % !!! FIXME
   numlist(Min,Max,NNs),
   dataset_histogram(CID, dense_pitch_hist(Min,Max,W), _{values:NNs}, Result).

collection_pitch_lookup(Weighting, Pitch, Lim, Offset, CID, Result) :-
   map_reduce_dataset(rec_pitch_hist(Weighting), pitch_lookup_cont(Pitch,Lim,Offset), CID, Result).

pitch_lookup_cont(Pitch,Lim,Offset,RecHists, _{items:Items}) :-
   findall( _{ uri: Rec, label:Label, count:Count, prob:Prob },
            offset(Offset, limit(Lim, order_by( [desc(Prob)], 
                   (  member(Rec-Hist,RecHists),
                      rdf_display_label(Rec,Label),
                      pitch_hist_prob(Hist,Pitch,Count,Prob)
                   )))),
            Items).

% collection_pitch_lookup_alt(Weighting, Pitch, Lim, Offset, CID, _{ items:Items, coverage:Coverage}) :-
%    findall_map_coverage(dataset_item(CID), rec_transcription, RecTrans, Coverage),
%    findall( _{ uri: Rec, label:Label, count:Count, prob:Prob },
%             offset(Offset, limit(Lim, order_by( [desc(Prob)],
%                    (  csv_pitch_count_prob(Weighting,Trans,Pitch,Count,Prob),
%                       member(Rec-Trans,RecTrans),
%                       rdf_display_label(Rec,Label)
%                    )))),
%             Items).
%
% rec_transcription(Rec,Rec-Transcription) :- transform_computation(transcription,Rec,Transcription).

collection_freq_histogram(Lang,Min,Max,Q,W,CID,Result) :-
   Map=binmap(Min,Max,(Max-Min)*Q+1),
   map_edges(Lang,Map,Edges),
   dataset_histogram(CID, dense_freq_hist(Lang,Map,W),_{edges:Edges}, Result).

collection_tempo_histogram(Lang,DT,Min,Max,N,CID,Result) :-
   insist(Min>0, domain_error(min,"positive value",Min)), 
   Map=expmap(Min,Max,N),
   map_edges(Lang,Map,Edges),
   dataset_histogram(CID, tempo_hist(Lang,DT,Map), _{edges:Edges}, Result).

collection_tempo_curve(Lang,N,CID, Result) :-
   map_reduce_dataset(tempo_curve(Lang,N), tempo_curves_stats(Lang), CID, Result).


dataset_histogram(CID, Mapper, Dict, Result) :-
   dataset_map_fold_reduce(CID,Mapper,with_dl(fold_hist),finish_hist(Dict),nothing,Result).
              
fold_hist([], S, S) :- !.
fold_hist(Xs, just(C1), just(C2)) :- !, insist(seqmap(maplist(add),Xs,C1,C2)).
fold_hist([X|Xs], nothing, just(C)) :- insist(seqmap(maplist(add),Xs,X,C)).

finish_hist(Dict,just(Counts),Hist) :- put_dict(counts,Dict,Counts,Hist).

py_hist(Mapper, PyFunction, Args, CID, _{counts:H,values:D,coverage:C,py_coverage:PYC}) :-
   py_cla(Mapper,PyFunction,Args, CID, _{stats:_{counts:H,domain:D},coverage:C,py_coverage:PYC}).

py_cla(Mapper, PyFunction, Args, CID, Result) :-
   map_reduce_dataset(Mapper, py_cla_cont(PyFunction,Args), CID, Result).

py_cla_cont(PyFunction,Args, Ok, _{stats:Result, py_coverage:Coverage}) :-
   python_apply(PyFunction,[Ok|Args],Reply),
   Reply = _{result:Result, stats:Coverage}.


% CLA mappers
rec_pitch_hist(W,Rec,Rec-Hist) :- transform_op(transcription,pitch_hist(W),Rec,Hist).

dense_pitch_hist(Min,Max,W,Rec,DenseHist) :-
   transform_op(transcription,pitch_hist(W),Rec,SparseHist),
   sparse_to_dense(Min,Max,SparseHist,DenseHist).

dense_freq_hist(ml,Map,W,Rec,Counts) :-
   transform_op(transcription(1),freq_hist(Map,W),Rec,Counts).
dense_freq_hist(r,Map,W,Rec,Counts) :-
   transform_op(transcription(1),freq_hist_r(Map,W),Rec,Counts).

tempo_hist(ml,DT,Map,Rec,Counts) :-
   transform_op(tempo,tempo_hist(DT,Map),Rec,Result),
   Result=_-Counts.
tempo_hist(r,DT,Map,Rec,Counts) :-
   transform_op(tempo,tempo_hist_r(DT,Map),Rec,Result),
   Result=_-Counts.

tempo_curve(ml,N,Rec,Values) :-
   transform_op(tempo,normalised_tempo(N),Rec,Result),
   Result=_-Values.
tempo_curve(r,N,Rec,Values) :-
   transform_op(tempo,normalised_tempo_r(N),Rec,Result),
   Result=_-Values.

transcription_tonic_duration(Rec, _{transcription: Transcription, tonic: Tonic, duration:0 }) :-
   tagged(transcription,Rec,Transcription),
   tagged(tonic,Rec,Tonic).

transcription_date(Rec, _{transcription: Transcription, date:Date}) :-
   tagged(transcription(1),Rec,Transcription),
   insist(recording_property(Rec,date,Date),missing_property(Rec,date)).

keys_chords(Rec, _{keys: Keys, chords:Chords}) :-
   tagged(key,Rec,Keys),
   tagged(chords,Rec,Chords).
   
similarity_bundle(Rec, _{chromagram: Chromagram, mfcc:Mfcc, keys: Keys, chords:Chords, list:_{uri:Rec, label:Label}}) :-
   % nb_getval(vamp_on_demand,Vamp),
   % concurrent_maplist(tagged_parallel(Vamp,Rec),[chromagram,mfcc,key,chords],[Chromagram,Mfcc,Keys,Chords]),
   maplist(tagged,[chromagram,mfcc,key,chords],[Rec,Rec,Rec,Rec],[Chromagram,Mfcc,Keys,Chords]),
   insist(recording_property(Rec,label,Label),missing_property(Rec,label)).
   
tagged_parallel(Vamp,Rec,Transform,Result) :-
   nb_setval(vamp_on_demand,Vamp),
   with_progress_stack(tagged(Transform,Rec,Result)).

list_places(Rec, _{place:Place,list:_{uri:Rec, label:Label}}) :-
   insist(recording_property(Rec,place,Place),missing_property(Rec,place)),
   insist(recording_property(Rec,label,Label),missing_property(Rec,label)).
   
% for later...
% tagged_list(Spec,Rec,Dict) :-
%    maplist(tagged_item(Rec),Spec,Pairs),
%    dict_create(Dict,_,Pairs).
% tagged_item(Rec,Key:Transform,Key:Value) :- tagged(Transform,Rec,Value).

tagged(Transform,Input,csv{value:Path}) :- 
   transform_computation(Transform,Input,R), uri_absolute_path(R,Path).

% ---------------------------------------------------

:- initialization time(memo_attach(memo(perspectives),[])).

:- persistent_memo cla_memo(+spec:ground,+cid:atom,-result:any).
cla_memo(Spec,CID,Result) :- 
   debug(perspectives(cla),'cla_mem: ~q',[call(Spec,CID,Result)]),
   with_progress_stack(call(Spec,CID,Result)).

%% mem(+Spec:pred(+cid,-A),+Opts:options,+CID:cid,-Result:A) is det.
%
%  Asynchronous memoised collection-level computation.
%  Spec must be a ground term that can be called with two arguments: the id of a
%  collection and a variable, which must be bound to an arbitrary result term on exit.
%  If the computation has already be done and memoised in cla_memo/3, then the result is
%  retrieved. Otherwise, the computation is started asynchronously and an exception 
%  describing the state of the computation will be thrown.
%  ==
%  * dml_error(10, _{status:already_waiting, position:n})
%     the goal was previously added and is now waiting at position n in the queue.
%  * dml_error(11, _{status:already_running, progress:Progress})
%     the goal was previously added and is currently running, with some progress information.
%  * dml_error(12, _{status:initiate, position:N})
%     Means the goal has been added to the work queue of the thread pool at position N.
%  ==
%  Options are options passed to control interaction with async_memo.

mem(Spec,Opts,CID,Result) :-
   option(vamp_on_demand(V), Opts, false),
   async_memo(vis_cla,cla_memo(Spec,CID,Result),Status,
              [ progress_levels([elapsed,summary,partial_result]),
                globals([vamp_on_demand-V])|Opts ]),
   (  Status=done(_-ok) -> true
   ;  status_response(Status,Code,Dict), 
      (  Status=done(_) -> Dict1=Dict
      ;  estimate_run_time(Spec,CID,ERT),
         put_dict(ert,Dict,ERT,Dict1)
      ),
      throw(dml_error(Code,Dict1))
   ).

% very crude estimate
estimate_run_time(Spec,CID0,ERT) :-
   findall(Size-Dur, (  browse(cla_memo(Spec,CID,_),comp(_,_,Dur)-ok),
                       dataset_size(CID,Size)), Pairs),
   length(Pairs,N),
   (  N=0 -> ERT is -1
   ;  maplist(computations:pair,Sizes,Durs,Pairs),
      sumlist(Sizes,TotalSize),
      sumlist(Durs,TotalDur),
      dataset_size(CID0,Size), 
      ERT is TotalDur*Size/TotalSize
   ).

status_response(spawned(ID,Pos), 12, Info) :-
   Info = _{status:initiated, id:ID, position:Pos }.
status_response(waiting(ID,T,Pos), 10, Info) :-
   Info = _{status:already_waiting, id:ID, submit_time:TS, position:Pos },
   time_to_string(T,TS).
status_response(running(ID,TStart,_,nothing), 11, Info) :-
   Info = _{ id:ID, status:already_running, start_time:TS },
   time_to_string(TStart,TS).
status_response(running(ID,TStart,_,just(Time-[Progress,Partial])), 11, Info) :-
   maplist(progress_json,Progress,Progs),
   time_to_string(TStart,TS),
   Elapsed is Time-TStart,
   (  member(stepwise(_,Done/Total),Progress), Done>0
   -> ETA is Elapsed*(Total-Done)/Done
   ;  ETA is -1
   ),
   Info1 = _{ id:ID
            , status:already_running
            , start_time:TS
            , elapsed_time:Elapsed
            , progress:Progs
            , eta:ETA
            },
   (  Partial=just(R) 
   -> put_dict(partial_result,Info1,R,Info)
   ;  Info=Info1
   ).

status_response(recomputing(ID,Pos,Meta), 13, Info) :-
   Info = _{status:recomputing, id:ID, position:Pos, meta:MD}, 
   meta_dict(Meta,MD).
status_response(done(Meta), 14, _{status:failed, meta:MD}) :- 
   meta_dict(Meta,MD).

meta_dict(comp(_,Time,Dur)-Result, _{ date: Date, duration:Dur, reason:Reason}) :-
   format_result(Result,Reason),
   time_to_string(Time,Date).

format_result(fail,'Unspecified failure').
format_result(ex(Ex),Description) :- message_to_string(Ex,Description).

time_to_string(Time,String) :- format_time(string(String),'%FT%T%:z',Time).

progress_json(A,A) :- atomic(A), !.
progress_json(stepwise(Desc,Done/Total), _{ task:Task, total:Total, done:Done }) :- !,
   progress_json(Desc,Task).
progress_json(T,A) :- message_to_string(T,A).

prolog:message(map_fold(_Mapper,_Folder)) --> ['Map-fold'].

:- multifile thread_pool:create_pool/1.
thread_pool:create_pool(vis_cla) :-
   current_prolog_flag(cpu_count,N),
   thread_pool_create(vis_cla, N, [backlog(20)]).

% ------------ computations with progress ------------------

map_reduce_dataset(Mapper,Reducer,CID,Result) :-
   dataset_map_fold_reduce(CID,Mapper,append_dl,with_dl(Reducer),H-H,Result).

append_dl(HH-TT,H-HH,H-TT).
with_dl(P,H-[],A) :- call(P,H,A).
with_dl(P,H-[],A,B) :- call(P,H,A,B).

dataset_map_fold_reduce(CID,Mapper,Folder,Reducer,S0,Result) :-
   dataset_items(CID,Items), 
   with_cont( 'Map-fold-reduce',
              map_fold_with_progress( safe_call(Mapper),
                                      safe_fold(Folder),
                                      Items, s(0,F-F,E-E,S0)),
              reduce_cont(Reducer), Result).

reduce_cont(Reducer,s(NOk,Failed-[],Erroneous-[],S), R) :-
   (  NOk>0
   -> simple_task(reducing(Reducer),call(Reducer,S,R1)),
      put_coverage(NOk,Failed,Erroneous,R1,R)
   ;  put_coverage(NOk,Failed,Erroneous,_{status:'no successfully mapped items'},D),
      throw(dml_error(20, D))
   ).

put_coverage(NOk,Failed,Erroneous,R1,R) :-
   put_dict(coverage,R1,_{ok_count:NOk, failed:Failed, errors:Erroneous},R).

safe_call(Mapper,X,Z) :-
   (  catch((call(Mapper,X,Y), Z=ok(X,Y)), Ex, 
            (  Ex=abort(_) -> throw(Ex)
            ;  Z=error(X,Ex))), !
   ;  Z=fail(X)
   ).

safe_fold(Folder,Items,s(NOk1,FH-FT1,EH-ET1,S1),s(NOk2,FH-FT2,EH-ET2,S2)) :-
   seqmap(partition,Items,s(NOk1,OkH,FT1,ET1),s(NOk2,OkT,FT2,ET2)),
   call(Folder,OkH-OkT,S1,S2).

partition(ok(_,X),s(N,[X|O],F,E),s(M,O,F,E)) :- M is N+1.
partition(fail(X),s(N,O,[X|F],E),s(N,O,F,E)).
partition(error(X,Ex),s(N,O,F,[_{item:X, error:Msg}|E]),s(N,O,F,E)) :- message_to_string(Ex,Msg).