<gh_stars>0
package com.stc.trawl.voice.job.compare;

import com.stc.trawl.common.core.exception.TrawlException;
import com.stc.trawl.common.pool.Pool;
import com.stc.trawl.common.rest.response.ResultStatus;
import com.stc.trawl.voice.data.compare.EtalonComparison;
import com.stc.trawl.voice.data.compare.TargetComparison;
import com.stc.trawl.voice.data.model.VoiceModelWithBinData;
import com.stc.trawl.voice.engine.model.VoiceModelEngine;
import com.stc.trawl.voice.job.compare.component.DataResolveAgent;
import com.stc.trawl.voice.job.compare.component.ModelResolveRequest;
import com.stc.trawl.voice.job.compare.component.ModelsResolveRequest;
import com.stc.trawl.voice.job.compare.component.VoiceModelResolveAgent;
import com.stc.trawl.voice.job.compare.v2.CompareResult;
import com.stc.trawl.voice.job.compare.v2.CompareResultItem;
import com.stc.trawl.voice.job.compare.v2.FilteredCompareResultItem;
import com.stc.trawl.voice.job.compare.v2.LimitedCompareResultItem;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.resources.SpringResource;

import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

/**
 * @author belyaev-a
 * @date 13.12.2019
 */
@Slf4j
@Data
@NoArgsConstructor
public abstract class AbstractCompareJob implements IgniteCallable<CompareResult> {
    private static final long serialVersionUID = 5344817269430279352L;

    private Collection<URI> targetUris;
    private Collection<URI> etalonUris;
    private Integer limit;
    private Predicate<EtalonComparison> filter;
    private String cacheName;

    protected Collection<TargetComparison> prepareTargetErrors;
    protected Collection<EtalonComparison> prepareEtalonErrors;

    {
        prepareTargetErrors = new ArrayList<>();
        prepareEtalonErrors = new ArrayList<>();
    }

    @SpringResource(resourceName = "voiceModelEnginePool")
    protected transient Pool<VoiceModelEngine> voiceModelEnginePool;

    public AbstractCompareJob(Collection<URI> targetUris, Collection<URI> etalonUris) {
        this.targetUris = targetUris;
        this.etalonUris = etalonUris;
    }

    private Function<URI, CompareResultItem> makeFunction() {
        Function<URI, CompareResultItem> resultItemFunction = uri -> new CompareResultItem(new TargetComparison(uri));

        if (limit != null)
            resultItemFunction = resultItemFunction.andThen(compareResultItem -> new LimitedCompareResultItem(compareResultItem, limit));

        if (filter != null)
            resultItemFunction = resultItemFunction.andThen(compareResultItem -> new FilteredCompareResultItem(compareResultItem, filter));

        return resultItemFunction;
    }

    protected Map<URI, VoiceModelWithBinData> prepareTargets() {
        DataResolveAgent<VoiceModelWithBinData> resolveAgent = new VoiceModelResolveAgent(this::getVoiceModelWithBinData);
        DataResolveAgent.Result<VoiceModelWithBinData> resolve = resolveAgent.resolve(
                ModelsResolveRequest.builder()
                        .cacheName(null)
                        .uris(targetUris)
                        .build());

        prepareTargetErrors = resolve.getExceptions().entrySet().stream()
                .map(uriErrorEntry -> {
                    URI key = uriErrorEntry.getKey();
                    DataResolveAgent.Result.Error value = uriErrorEntry.getValue();
                    return new TargetComparison(key, ResultStatus.ERROR, value.getCode(), value.getDescription());
                })
                .collect(Collectors.toList());
        return resolve.getUriDataMap();
    }

    protected Map<URI, VoiceModelWithBinData> prepareEtalons() throws TrawlException {
        DataResolveAgent<VoiceModelWithBinData> resolveAgent = new VoiceModelResolveAgent(this::getVoiceModelWithBinData);
        DataResolveAgent.Result<VoiceModelWithBinData> resolve = resolveAgent.resolve(ModelsResolveRequest.builder()
                .cacheName(cacheName)
                .uris(etalonUris)
                .build());

        prepareEtalonErrors = resolve.getExceptions().entrySet().stream()
                .map(uriErrorEntry -> {
                    URI key = uriErrorEntry.getKey();
                    DataResolveAgent.Result.Error value = uriErrorEntry.getValue();
                    return EtalonComparison.error(key, value.getCode(), value.getDescription());
                })
                .collect(Collectors.toList());
        return resolve.getUriDataMap();
    }

    protected abstract VoiceModelWithBinData getVoiceModelWithBinData(ModelResolveRequest request) throws Exception;

    @Override
    public CompareResult call() throws Exception {
        Map<URI, VoiceModelWithBinData> uriTargetMap = prepareTargets();
        Map<URI, VoiceModelWithBinData> uriEtalonMap = prepareEtalons();
        log.debug("Total compare count: {}", uriTargetMap.size() * uriEtalonMap.size());

        Function<URI, CompareResultItem> compareResultItemFunction = makeFunction();

        VoiceModelEngine voiceModelEngine = voiceModelEnginePool.borrow();
        try {
            ArrayList<CompareResultItem> comparisons = new ArrayList<>(targetUris.size());
            for (Map.Entry<URI, VoiceModelWithBinData> targetEntry : uriTargetMap.entrySet()) {
                URI targetUri = targetEntry.getKey();
                VoiceModelWithBinData targetModel = targetEntry.getValue();
                CompareResultItem compareResultItem = compareResultItemFunction.apply(targetUri);

                List<EtalonComparison> etalonComparisons = voiceModelEngine.compareVSMap(targetModel, uriEtalonMap).stream()
                        .map(engineCompareResult -> {
                            if (engineCompareResult.isError()) {
                                return EtalonComparison.error(
                                        engineCompareResult.getEtalon(),
                                        engineCompareResult.getCode(),
                                        engineCompareResult.getDescription());
                            } else {
                                return EtalonComparison.ok(
                                        engineCompareResult.getEtalon(),
                                        engineCompareResult.getRelevance(),
                                        engineCompareResult.getFr(),
                                        engineCompareResult.getFa());
                            }
                        }).collect(Collectors.toList());

                etalonComparisons.forEach(compareResultItem::addResult);
                compareResultItem.addErrors(prepareEtalonErrors);
                comparisons.add(compareResultItem);
            }
            comparisons.addAll(prepareTargetErrors.stream()
                    .map(CompareResultItem::new)
                    .collect(Collectors.toList()));
            return new CompareResult(comparisons);
        } finally {
            voiceModelEnginePool.retrieve(voiceModelEngine);
        }
    }

}
