ProcessRanking.java

Go to the documentation of this file.

/*
 * Turró i Cutiller Foundation. License notice.
 * Copyright (C) 2020 Lluis Turró Cutiller <http://www.turro.org/>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
package org.turro.ranking;
 
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Stream;
import org.turro.action.IProcess;
import org.turro.elephant.context.ElephantContext;
import org.turro.elephant.db.WhereClause;
import org.turro.indicator.Formulas;
import org.turro.indicator.IElephantIndicator;
import org.turro.indicator.IPreprocessor;
import org.turro.indicator.IndicatorVariable;
import org.turro.indicator.Indicators;
import org.turro.indicator.VariableType;
import org.turro.jpa.Dao;
import org.turro.math.Zero;
 
public abstract class ProcessRanking<T> implements IProcess, IPreprocessor {
 
  @Override
  public void startProcess() {
    Logger.getLogger(ProcessRanking.class.getName())
            .log(Level.INFO, ElephantContext.logMsg("Start " + instanceClass().getSimpleName() + "..."));
    start();
    truncate();
    Logger.getLogger(ProcessRanking.class.getName())
            .log(Level.INFO, ElephantContext.logMsg("Truncated"));
    List<IndicatorVariable> vars = Indicators.getRankingVariables(indicatorsRoot());
    prepareProcess(vars);
    Logger.getLogger(ProcessRanking.class.getName())
            .log(Level.INFO, ElephantContext.logMsg("Preprocessed"));
    if(Indicators.hasDirectProcess(vars)) {
      try(Stream<T> entities = getEntityStream()) {
        process(entities, vars);
      }
    }
    Logger.getLogger(ProcessRanking.class.getName())
            .log(Level.INFO, ElephantContext.logMsg("Processed"));
    postProcess(vars);
    Logger.getLogger(ProcessRanking.class.getName())
            .log(Level.INFO, ElephantContext.logMsg("Postprocessed"));
    calculateRanking();
    end();
    Logger.getLogger(ProcessRanking.class.getName())
            .log(Level.INFO, ElephantContext.logMsg("End " + instanceClass().getSimpleName()));
  }
 
  @Override
  public void startProcessFor(Object entity) {
    String entityPath = getDao().getPath(entity);
    WhereClause wc = new WhereClause();
    wc.addClause("delete from " + instanceClass().getSimpleName());
    wc.addClause("where entityPath = :entityPath");
    wc.addNamedValue("entityPath", entityPath);
    getDao().executeUpdate(wc);
    processFor((T) entity, Indicators.getRankingVariables(indicatorsRoot()));
  }
 
  public void process(Stream<T> stream, List<IndicatorVariable> variables) {
    try {
      stream.forEach(t -> {
        processFor(t, variables);
      });
    } catch(Exception ex) {
      Logger.getLogger(ProcessRanking.class.getName())
            .log(Level.SEVERE, ElephantContext.logMsg("Processing"), ex);
    }
  }
  
  public void processFor(T entity, List<IndicatorVariable> variables) {
    prepareEntity(entity);
    variables.forEach(v -> {
      if(!v.isPreprocess()) {
        process(entity, v);
      }
    });
  }
 
  protected void start() {}
  protected void end() {}
  
  protected void truncate() {
    getDao().executeNativeUpdate("truncate table " + instanceClass().getSimpleName());
  }
  
  protected IRanking process(T entity, IndicatorVariable variable) {
    IRanking ranking = createRankingInstance(getDao().getPath(entity), variable);
    ranking.setRanking(Indicators.getValue(entity, variable));
    if(!Zero.near(ranking.getRanking(), 2)) {
      ranking = getDao().saveObject(ranking);
    }
    return ranking;
  }
 
  protected void prepareProcess(List<IndicatorVariable> variables) {
    variables.forEach(v -> {
      if(v.isPreprocess()) {
        IElephantIndicator ei = Indicators.getIndicator(v);
        ei.preprocess(v, this, entitiesRoot(), null);
      }
    });
  }
  
  protected void postProcess(List<IndicatorVariable> variables) {
    variables.forEach(v -> {
      IElephantIndicator ei = Indicators.getIndicator(v);
      ei.postprocess(v, this, entitiesRoot(), null);
    });
  }
  
  protected void prepareEntity(T entity) { }
  
  protected double calculate(RankingSet set) {
    return set.applyFrom(Formulas.load().getFormula(instanceClass()));
  }
 
  protected IRanking createRankingInstance(String entityPath, IndicatorVariable variable) {
    IRanking ranking = createRankingInstance();
    ranking.setEntityPath(entityPath);
    ranking.setConcept(variable.getName());
    ranking.setRanking(0);
    return ranking;
  }
 
  protected abstract Stream<T> getEntityStream();
  protected abstract IRanking createRankingInstance();
  protected abstract Class instanceClass();
  protected abstract String indicatorsRoot();
  protected abstract String entitiesRoot();
  protected abstract Dao createDao();
  
  /* Results */
  
  private IRanking previous;
  private RankingSet set;
  private List<IndicatorVariable> variables;
 
  private void calculateRanking() {
    previous = null;
    set = new RankingSet();
    variables = Indicators.getRankingVariables(indicatorsRoot());
    try(Stream<IRanking> tuples = getIndicatorTuples()) {
      tuples.forEach(t -> doCalculate(t));
    }
    if(!set.isEmpty()) {
      doCalculate(null);
    }
    getDao().finishPool();
  }
  
  private Stream<IRanking> getIndicatorTuples() {
    WhereClause wc = new WhereClause();
    wc.addClause("select m from " + instanceClass().getSimpleName() + " m");
    wc.addClause("order by entityPath, concept");
    return createDao().stream(IRanking.class, wc, 1000);  
  }
  
  private void doCalculate(IRanking ranking) {
    if(ranking == null || (previous != null && !sameTuple(previous, ranking))) {
      set.addMissing(variables);
      IRanking result = createRankingInstance(previous.getEntityPath(), 
              IndicatorVariable.result(indicatorsRoot(), VariableType.RANKING_VARIABLE));
      result.setRanking(calculate(set));
      if(!Zero.near(result.getRanking(), 2)) {
        getDao().poolObject(result, 1000);
      }
      set = new RankingSet();
    }
    if(ranking != null) {
      set.add(ranking);
      previous = ranking;
    }
  }
 
  private boolean sameTuple(IRanking ranking1, IRanking ranking2) {
    if(ranking1 == null || ranking2 == null) {
      return false;
    }
    return ranking1.getEntityPath().equals(ranking2.getEntityPath());
  }
  
  /* Preprocessor */
 
  @Override
  public Object createInstance() {
    return createRankingInstance();
  }
 
  @Override
  public void poolInstance(Object value) {
    getDao().poolObject(value, 1000);
  }
 
  @Override
  public void finishPreprocessor() {
    getDao().finishPool();
  }
  
  @Override
  public String tableName() {
    return instanceClass().getSimpleName();
  }  
  
  /* Dao */
  
  private Dao _dao;
  
  protected Dao getDao() {
    if(_dao == null) {
      _dao = createDao();
    }
    return _dao;
  }
 
}