<commit_before>:- module(
  io,
  [
    call_on_stream/2,        % +Source, :Goal_3
    call_on_stream/3,        % +Source, :Goal_3, +SourceOpts
    call_onto_stream/3,      % +Source, +Sink, :Goal_4
    call_onto_stream/5,      % +Source, +Sink, :Goal_4, +SourceOpts, +SinkOpts
    call_onto_streams/4,     % +Source, +Sink1, +Sink2, :Goal_5
    call_onto_streams/6,     % +Source, +Sink1, +Sink2, :Goal_5, +SourceOpts, +SinkOpts
    call_onto_streams/7,     % +Source, +Sink1, +Sink2, :Goal_5, +SourceOpts, +Sink1Opts, +Sink2Opts
    call_to_atom/2,          % :Goal_1, -A
    call_to_codes/2,         % :Goal_1, -Cs
    call_to_stream/2,        % +Sink, :Goal_1
    call_to_stream/3,        % +Sink, :Goal_1, +SinkOpts
    call_to_streams/3,       % +Sink1, +Sink2, :Goal_2
    call_to_streams/4,       % +Sink1, +Sink2, :Goal_2, +SinkOpts
    call_to_streams/5,       % +Sink1, +Sink2, :Goal_2, +Sink1Opts, +Sink2Opts
    call_to_string/2,        % :Goal_1, -Str
    close_any/2,             % +Close, -Meta
    close_any/3,             % +Close, +Meta1, -Meta2
    read_line_to_atom/2,     % +In, -A
    read_mode/1,             % ?Mode
    read_stream_to_atom/2,   % +In, -A
    read_stream_to_string/2, % +In, -Str
    stream_metadata/2,       % +Stream, -Meta
    write_mode/1             % ?Mode
  ]
).

/** <module> I/O

@author <NAME>
@tbd Implement metadata using backtrackable setval.
@version 2016/07
*/

:- use_module(library(iostream)).
:- use_module(library(os/archive_ext)).
:- use_module(library(zlib)).

:- meta_predicate
    call_on_stream(+, 3),
    call_on_stream(+, 3, +),
    call_on_stream0(+, +, 3, +, -, +),
    call_onto_stream(+, +, 4),
    call_onto_stream(+, +, 4, +, +),
    call_onto_stream0(+, 4, +, +, +, -),
    call_onto_streams(+, +, +, 5),
    call_onto_streams(+, +, +, 5, +, +),
    call_onto_streams(+, +, +, 5, +, +, +),
    call_onto_streams0(+, +, 5, +, +, +, +, -),
    call_to_atom(1, -),
    call_to_codes(1, -),
    call_to_stream(+, 1),
    call_to_stream(+, 1, +),
    call_to_streams(+, +, 2),
    call_to_streams(+, +, 2, +),
    call_to_streams(+, +, 2, +, +),
    call_to_string(1, +).





%! call_on_stream(+Source, :Goal_3) is det.
%! call_on_stream(+Source, :Goal_3, +SourceOpts) is det.
%
% The following call is made: `call(Goal_3, In)`.
%
% The following options are supported:
%
%   * entry_name(?atom)
%
%   * metadata(-dict)
%
%   * Other options are passed to:
%
%       * archive_data_stream/3
%       * archive_open/3
%       * open_any/5

call_on_stream(Source, Goal_3) :-
  call_on_stream(Source, Goal_3, []).


call_on_stream(Source, Goal_3, SourceOpts) :-
  % This allows the calling context to request one specific entity.
  option(entry_name(EntryName), SourceOpts, _),
  % Archive options that cannot be overridden.
  findall(format(Format), archive_format(Format, true), FormatOpts),
  merge_options([close_parent(false)|FormatOpts], SourceOpts, ArchOpts1),
  % Archive options that can be overridden.
  merge_options(ArchOpts1, [filter(all)], ArchOpts2),
  merge_options([metadata(Meta1)], SourceOpts, OpenOpts),
  setup_call_cleanup(
    open_any(Source, read, In1, Close, OpenOpts),
    setup_call_cleanup(
      archive_open(In1, Arch, ArchOpts2),
      (
        (var(Meta1) -> Meta1 = _{} ; true),
        (   source_base_iri(Source, BaseIri)
        ->  put_dict(base_iri, Meta1, BaseIri, Meta2)
        ;   Meta2 = Meta1
        ),
        % Semi-deterministic if an archive entry name is given.
        (   nonvar(EntryName)
        ->  once(call_on_stream0(Arch, EntryName, Goal_3, Meta2, Meta3, SourceOpts))
        ;   call_on_stream0(Arch, EntryName, Goal_3, Meta2, Meta3, SourceOpts)
        )
      ),
      archive_close(Arch)
    ),
    close_any(Close, Meta3, Meta4)
  ),
  ignore(option(metadata(Meta4), SourceOpts)).


call_on_stream0(Arch, EntryName, Goal_3, Meta1, Meta4, SourceOpts1) :-
  merge_options([meta_data(Path)], SourceOpts1, SourceOpts2),
  archive_data_stream(Arch, In2, SourceOpts2),
  % Even though the first input stream `In1` used UTF-8 encoding, the
  % streams that come out an any archive are `octet`.  This is
  % corrected here.
  set_stream(In2, encoding(utf8)),
  (   Path = [Entry|_],
      EntryName = Entry.name
  ->  put_dict(path, Meta1, Path, Meta2),
      call_cleanup(
        call(Goal_3, In2, Meta2, Meta3),
        close_any(close(In2), Meta3, Meta4)
      )
  ;   close(In2),
      fail
  ).



%! call_onto_stream(+Source, +Sink, :Goal_4) is det.
%! call_onto_stream(+Source, +Sink, :Goal_4, +SourceOpts, +SinkOpts) is det.
%<commit_msg>ENHANCED: Remove the last entry from the archive entry path, which is always data/raw @io<commit_after>:- module(
  io,
  [
    call_on_stream/2,        % +Source, :Goal_3
    call_on_stream/3,        % +Source, :Goal_3, +SourceOpts
    call_onto_stream/3,      % +Source, +Sink, :Goal_4
    call_onto_stream/5,      % +Source, +Sink, :Goal_4, +SourceOpts, +SinkOpts
    call_onto_streams/4,     % +Source, +Sink1, +Sink2, :Goal_5
    call_onto_streams/6,     % +Source, +Sink1, +Sink2, :Goal_5, +SourceOpts, +SinkOpts
    call_onto_streams/7,     % +Source, +Sink1, +Sink2, :Goal_5, +SourceOpts, +Sink1Opts, +Sink2Opts
    call_to_atom/2,          % :Goal_1, -A
    call_to_codes/2,         % :Goal_1, -Cs
    call_to_stream/2,        % +Sink, :Goal_1
    call_to_stream/3,        % +Sink, :Goal_1, +SinkOpts
    call_to_streams/3,       % +Sink1, +Sink2, :Goal_2
    call_to_streams/4,       % +Sink1, +Sink2, :Goal_2, +SinkOpts
    call_to_streams/5,       % +Sink1, +Sink2, :Goal_2, +Sink1Opts, +Sink2Opts
    call_to_string/2,        % :Goal_1, -Str
    close_any/2,             % +Close, -Meta
    close_any/3,             % +Close, +Meta1, -Meta2
    read_line_to_atom/2,     % +In, -A
    read_mode/1,             % ?Mode
    read_stream_to_atom/2,   % +In, -A
    read_stream_to_string/2, % +In, -Str
    stream_metadata/2,       % +Stream, -Meta
    write_mode/1             % ?Mode
  ]
).

/** <module> I/O

@author <NAME>
@tbd Implement metadata using backtrackable setval.
@version 2016/07
*/

:- use_module(library(iostream)).
:- use_module(library(os/archive_ext)).
:- use_module(library(zlib)).

:- meta_predicate
    call_on_stream(+, 3),
    call_on_stream(+, 3, +),
    call_on_stream0(+, +, 3, +, -, +),
    call_onto_stream(+, +, 4),
    call_onto_stream(+, +, 4, +, +),
    call_onto_stream0(+, 4, +, +, +, -),
    call_onto_streams(+, +, +, 5),
    call_onto_streams(+, +, +, 5, +, +),
    call_onto_streams(+, +, +, 5, +, +, +),
    call_onto_streams0(+, +, 5, +, +, +, +, -),
    call_to_atom(1, -),
    call_to_codes(1, -),
    call_to_stream(+, 1),
    call_to_stream(+, 1, +),
    call_to_streams(+, +, 2),
    call_to_streams(+, +, 2, +),
    call_to_streams(+, +, 2, +, +),
    call_to_string(1, +).





%! call_on_stream(+Source, :Goal_3) is det.
%! call_on_stream(+Source, :Goal_3, +SourceOpts) is det.
%
% The following call is made: `call(Goal_3, In)`.
%
% The following options are supported:
%
%   * entry_name(?atom)
%
%   * metadata(-dict)
%
%   * Other options are passed to:
%
%       * archive_data_stream/3
%       * archive_open/3
%       * open_any/5

call_on_stream(Source, Goal_3) :-
  call_on_stream(Source, Goal_3, []).


call_on_stream(Source, Goal_3, SourceOpts) :-
  % This allows the calling context to request one specific entity.
  option(entry_name(EntryName), SourceOpts, _),
  % Archive options that cannot be overridden.
  findall(format(Format), archive_format(Format, true), FormatOpts),
  merge_options([close_parent(false)|FormatOpts], SourceOpts, ArchOpts1),
  % Archive options that can be overridden.
  merge_options(ArchOpts1, [filter(all)], ArchOpts2),
  merge_options([metadata(Meta1)], SourceOpts, OpenOpts),
  setup_call_cleanup(
    open_any(Source, read, In1, Close, OpenOpts),
    setup_call_cleanup(
      archive_open(In1, Arch, ArchOpts2),
      (
        (var(Meta1) -> Meta1 = _{} ; true),
        (   source_base_iri(Source, BaseIri)
        ->  put_dict(base_iri, Meta1, BaseIri, Meta2)
        ;   Meta2 = Meta1
        ),
        % Semi-deterministic if an archive entry name is given.
        (   nonvar(EntryName)
        ->  once(call_on_stream0(Arch, EntryName, Goal_3, Meta2, Meta3, SourceOpts))
        ;   call_on_stream0(Arch, EntryName, Goal_3, Meta2, Meta3, SourceOpts)
        )
      ),
      archive_close(Arch)
    ),
    close_any(Close, Meta3, Meta4)
  ),
  ignore(option(metadata(Meta4), SourceOpts)).


call_on_stream0(Arch, EntryName, Goal_3, Meta1, Meta4, SourceOpts1) :-
  merge_options([meta_data(Path0)], SourceOpts1, SourceOpts2),
  archive_data_stream(Arch, In2, SourceOpts2),
  % Even though the first input stream `In1` used UTF-8 encoding, the
  % streams that come out an any archive are `octet`.  This is
  % corrected here.
  append(Path, [_], Path0),
  set_stream(In2, encoding(utf8)),
  (   Path = [Entry|_],
      EntryName = Entry.name
  ->  put_dict(path, Meta1, Path, Meta2),
      call_cleanup(
        call(Goal_3, In2, Meta2, Meta3),
        close_any(close(In2), Meta3, Meta4)
      )
  ;   close(In2),
      fail
  ).



%! call_onto_stream(+Source, +Sink, :Goal_4) is det.
%! call_onto_stream(+Source, +Sink, :Goal_4, +SourceOpts, +SinkOpts) is det.
%
{'fraction_non_alphanumeric': 0.1542869499405341, 'fraction_numerical': 0.030706022272678126, 'mean_word_length': 3.0552389302937306, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2045094', 'n_tokens_mistral': 3715, 'n_tokens_neox': 3708, 'n_words': 686}