<commit_before>-module(dqe_aggr2).

-behaviour(dflow).

-export([init/1, describe/1, start/2, emit/3, done/2]).

-record(state, {
          aggr :: atom(),
          time :: pos_integer(),
          acc = <<>> :: binary(),
          resolution :: pos_integer()
         }).

init([Aggr, SubQ, Time]) ->
    {ok, #state{aggr = Aggr, time = Time}, SubQ}.

start({_Start, _Count}, State) ->
    {ok, State}.

describe(#state{aggr = Aggr, time = Time}) ->
    [atom_to_list(Aggr), "(", integer_to_list(round(Time/1000)), "ms)"].

%% When we get the first data we can calculate both the applied
%% time and the upwards resolution.
emit(Child, {Data, Resolution},
     State = #state{resolution = undefined, time = Time}) ->
    Time1 = dqe_time:apply_times(Time, Resolution),
    emit(Child, {Data, Resolution},
         State#state{resolution = Time1 * Resolution, time = Time1});

emit(_Child, {Data, _R},
     State = #state{aggr = Aggr, time = Time, acc = Acc}) ->
    case execute(Aggr, <<Data/binary, Acc/binary>>, Time, <<>>) of
        {Acc1, <<>>} ->
            {ok, State#state{acc = Acc1}};
        {Acc2, AccEmit} ->
            {emit, {AccEmit, State#state.resolution}, State#state{acc = Acc2}}
    end.


done(_Child, State = #state{acc = <<>>}) ->
    {done, State};

done(_Child, State = #state{aggr = Aggr, time = Time, acc = Acc}) ->
    Data = mmath_aggr:Aggr(Acc, Time),
    {done, {Data, State#state.resolution}, State#state{acc = <<>>}}.


execute(Aggr, Acc, T1, AccEmit) when byte_size(Acc) >= T1 * 9 ->
    MinSize = T1 * 9,
    <<Data:MinSize/binary, Acc1/binary>> = Acc,
    Result = mmath_aggr:Aggr(Data, T1),
    execute(Aggr, Acc1, T1, <<AccEmit/binary, Result/binary>>);

execute(_, Acc, _, AccEmit) ->
    {Acc, AccEmit}.<commit_msg>fixed bug with hardcoded data size.<commit_after>-module(dqe_aggr2).

-behaviour(dflow).

-include_lib("mmath/include/mmath.hrl").

-export([init/1, describe/1, start/2, emit/3, done/2]).

-record(state, {
          aggr :: atom(),
          time :: pos_integer(),
          acc = <<>> :: binary(),
          resolution :: pos_integer()
         }).

init([Aggr, SubQ, Time]) ->
    {ok, #state{aggr = Aggr, time = Time}, SubQ}.

start({_Start, _Count}, State) ->
    {ok, State}.

describe(#state{aggr = Aggr, time = Time}) ->
    [atom_to_list(Aggr), "(", integer_to_list(round(Time/1000)), "ms)"].

%% When we get the first data we can calculate both the applied
%% time and the upwards resolution.
emit(Child, {Data, Resolution},
     State = #state{resolution = undefined, time = Time}) ->
    Time1 = dqe_time:apply_times(Time, Resolution),
    emit(Child, {Data, Resolution},
         State#state{resolution = Time1 * Resolution, time = Time1});

emit(_Child, {Data, _R},
     State = #state{aggr = Aggr, time = Time, acc = Acc}) ->
    case execute(Aggr, <<Data/binary, Acc/binary>>, Time, <<>>) of
        {Acc1, <<>>} ->
            {ok, State#state{acc = Acc1}};
        {Acc2, AccEmit} ->
            {emit, {AccEmit, State#state.resolution}, State#state{acc = Acc2}}
    end.


done(_Child, State = #state{acc = <<>>}) ->
    {done, State};

done(_Child, State = #state{aggr = Aggr, time = Time, acc = Acc}) ->
    Data = mmath_aggr:Aggr(Acc, Time),
    {done, {Data, State#state.resolution}, State#state{acc = <<>>}}.


execute(Aggr, Acc, T1, AccEmit) when byte_size(Acc) >= T1 * 9 ->
    MinSize = T1 * ?DATA_SIZE,
    <<Data:MinSize/binary, Acc1/binary>> = Acc,
    Result = mmath_aggr:Aggr(Data, T1),
    execute(Aggr, Acc1, T1, <<AccEmit/binary, Result/binary>>);

execute(_, Acc, _, AccEmit) ->
    {Acc, AccEmit}.
{'fraction_non_alphanumeric': 0.19231839688282773, 'fraction_numerical': 0.014194266629557473, 'mean_word_length': 3.140552995391705, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 59, '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': '4580186', 'n_tokens_mistral': 1314, 'n_tokens_neox': 1281, 'n_words': 301}