kaa

-module(kaa_proto).

-export([worker/1,
    exec/1
]).

-include("kaa_core.hrl").
-include("kaa.hrl"). % auto-generated by gpb
-include("kaa_worker.hrl"). % auto-generated by gpb
-include("kaa_error.hrl"). % auto-generated by gpb
-include("kaa_result.hrl"). % auto-generated by gpb

worker(Pid) when not is_pid(Pid) ->
    KaaError = #'KaaError'{error = "no_jun_worker"},
    kaa_error:encode_msg(KaaError);
worker(Pid) when is_pid(Pid)     ->
    KaaWorker = #'KaaWorker'{jun_worker = pid_to_list(Pid)},
    kaa_worker:encode_msg(KaaWorker).

exec(PBMsg) when not is_binary(PBMsg) ->
    lager:error("invalid kaa protobuf message ~p", [invalid_binary]),
    KaaError = #'KaaError'{error = "invalid_kaa_proto_message"},
    kaa_error:encode_msg(KaaError);
exec(PBMsg) when is_binary(PBMsg)     ->
    case catch kaa:decode_msg(PBMsg, 'Kaa') of
        {'EXIT', Error}  ->
            lager:error("invalid kaa protobuf message ~p", [Error]),
            KaaError = #'KaaError'{error = "invalid_kaa_proto_message"},
            kaa_error:encode_msg(KaaError);
        #'Kaa'{} = KaaPB ->
            % decompose message and execute through jun!
            JunWorker = list_to_pid(KaaPB#'Kaa'.jun_worker),
            Mod = KaaPB#'Kaa'.module,
            Fn = KaaPB#'Kaa'.'function',
            Result = exec_jun(KaaPB#'Kaa'.arguments, JunWorker, Mod, Fn),
            encode_result(Result)
    end.

%% @hidden

exec_jun({core, #m_core{argument = Argument, keywords = Keywords}},
        JunWorker, Mod, Fn) ->
    Pid = pid_to_list(self()),
    % parse keywords in order to convert to a plist for jun
    Keywords0 = parse_keywords(Keywords),
    % maybe argument use a series, check for it in kaa environment
    Argument0 = parse_argument(Argument, Pid),
    Mod:Fn(JunWorker, Argument0, Keywords0);
exec_jun({frame, #m_frame{dataframe = MemId, axis = Axis, keywords = Keywords}},
        JunWorker, Mod, Fn) ->
    Pid = pid_to_list(self()),
    % lookup for a dataframe if is in mem, otherwise use original
    [{_, DataFrame}] = case ets:lookup(?KAA_ENVIRONMENT(Pid), MemId) of
      []    -> [{exclude, parse_argument(MemId, Pid)}];
      Found -> Found
    end,
    % parse keywords in order to convert to a plist for jun
    Keywords0 = parse_keywords(Keywords),
    % maybe dont use axis, this must be optional in proto
    case Axis of
        undefined ->
            Mod:Fn(JunWorker, DataFrame, Keywords0);
        _         ->
            % this rare condition happens when jun tries to use a integer value as
            % argument of a function, for example, head or tail.
            % a function could receive a float value?, if so, then parse.
            Axis0 = parse_argument(Axis, Pid),
            Mod:Fn(JunWorker, DataFrame, Axis0, Keywords0)
    end.

%% @hidden

encode_result({ok, {?SERIES, {'$erlport.opaque', python, _} = Series}})       ->
    MemId = random_key(),
    Pid = pid_to_list(self()),
    true = ets:insert(?KAA_ENVIRONMENT(Pid), {binary_to_list(MemId), Series}),
    KaaResult = #'KaaResult'{ok = "ok", result = {series, MemId}},
    kaa_result:encode_msg(KaaResult);
encode_result({ok, {?DATAFRAME, {'$erlport.opaque', python, _} = DataFrame}}) ->
    MemId = random_key(),
    Pid = pid_to_list(self()),
    true = ets:insert(?KAA_ENVIRONMENT(Pid), {binary_to_list(MemId), DataFrame}),
    KaaResult = #'KaaResult'{ok = "ok", result = {dataframe, MemId}},
    kaa_result:encode_msg(KaaResult);
encode_result({ok, {?GROUPBY, {'$erlport.opaque', python, _} = GroupBy}}) ->
    MemId = random_key(),
    Pid = pid_to_list(self()),
    true = ets:insert(?KAA_ENVIRONMENT(Pid), {binary_to_list(MemId), GroupBy}),
    KaaResult = #'KaaResult'{ok = "ok", result = {groupby, MemId}},
    kaa_result:encode_msg(KaaResult);
% the process of return plotting through pb is complex, since is an opaque term
% similar to dataframe, so maybe store in an internal storage to execute tasks
% in the plot after creation.
encode_result({ok, {?AXESPLOT, {'$erlport.opaque', python, _} = Plot}}) ->
    PlotBin = term_to_binary(Plot),
    KaaResult = #'KaaResult'{ok = "ok", result = {axesplot, binary_to_list(PlotBin)}},
    kaa_result:encode_msg(KaaResult);
encode_result({ok, {?SEABORNPLOT, {'$erlport.opaque', python, _} = Plot}}) ->
    PlotBin = term_to_binary(Plot),
    KaaResult = #'KaaResult'{ok = "ok", result = {seabornplot, binary_to_list(PlotBin)}},
    kaa_result:encode_msg(KaaResult);
encode_result({ok, {?PLOTLY, A}})                               ->
    PlotPickled = binary_to_list(A),
    KaaResult = #'KaaResult'{ok = "ok", result = {iplot, PlotPickled}},
    kaa_result:encode_msg(KaaResult);
encode_result({ok, I}) when is_integer(I)                       ->
    KaaResult = #'KaaResult'{ok = "ok", result = {inumber, I}},
    kaa_result:encode_msg(KaaResult);
encode_result({ok, F}) when is_float(F)                         ->
    KaaResult = #'KaaResult'{ok = "ok", result = {dnumber, F}},
    kaa_result:encode_msg(KaaResult);
encode_result({ok, B}) when is_binary(B)                        ->
    KaaResult = #'KaaResult'{ok = "ok", result = {string, binary_to_list(B)}},
    kaa_result:encode_msg(KaaResult);
encode_result({ok, S}) when is_list(S)                          ->
    KaaResult = #'KaaResult'{ok = "ok", result = {string, S}},
    kaa_result:encode_msg(KaaResult);
encode_result({error, {Error, Description}})                    ->
    KaaError = #'KaaError'{error = atom_to_list(Error),
        description = Description},
    kaa_error:encode_msg(KaaError).

%% @hidden

random_key() ->
    Seq = lists:seq(1, 100),
    Chars = "abcdeefghijklmnopqrstuvwxyz",
    R = lists:foldl(fun(_, Acc) ->
        L = length(Chars),
        [ lists:nth(rand:uniform(L), Chars) | Acc]
    end, [], Seq),
    list_to_binary(R).

%% @hidden

parse_keywords(Keywords) ->
    lists:map(fun(#'Keywords'{key = Key, value = Value}) ->
        case Value of
            {svalue, ValueStr} -> {list_to_binary(Key), list_to_binary(ValueStr)};
            {ivalue, ValueInt} -> {list_to_binary(Key), ValueInt};
            {dvalue, ValueDou} -> {list_to_binary(Key), ValueDou};
            {avalue, {'KeywordValue', ArrStr, [], []}} ->
                {list_to_binary(Key), normalize_array(ArrStr)};
            {avalue, {'KeywordValue', [], ArrInt, []}} ->
                {list_to_binary(Key), normalize_array(ArrInt)};
            {avalue, {'KeywordValue', [], [], ArrDou}} ->
                {list_to_binary(Key), normalize_array(ArrDou)}
        end
    end, Keywords).

%% @hidden

parse_argument(Argument, Pid) ->
    case catch list_to_integer(Argument) of
        {'EXIT', _} ->
            % if the argument is a series then check if we hold into kaa environment
            case ets:lookup(?KAA_ENVIRONMENT(Pid), Argument) of
                [{_, Series}] -> Series;
                _             -> list_to_binary(Argument)
            end;
        ArgumentInt -> ArgumentInt
    end.

%% @hidden

normalize_array([]) -> [];
normalize_array([ V | Array]) when is_list(V) ->
    [ list_to_binary(V) | normalize_array(Array) ];
normalize_array([ V | Array]) when is_integer(V) ->
    [ V | normalize_array(Array) ];
normalize_array([ V | Array]) when is_float(V) ->
    [ V | normalize_array(Array) ].