<commit_before>// Copyright © 2015 HSL <https://www.hsl.fi>
// This program is dual-licensed under the EUPL v1.2 and AGPLv3 licenses.

package fi.hsl.parkandride.core.service;

import fi.hsl.parkandride.core.back.PredictionRepository;
import fi.hsl.parkandride.core.back.PredictorRepository;
import fi.hsl.parkandride.core.back.UtilizationRepository;
import fi.hsl.parkandride.core.domain.Utilization;
import fi.hsl.parkandride.core.domain.UtilizationKey;
import fi.hsl.parkandride.core.domain.prediction.*;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.*;

import static org.springframework.transaction.annotation.Isolation.READ_COMMITTED;
import static org.springframework.transaction.annotation.Propagation.REQUIRES_NEW;

public class PredictionService {

    private static final Logger log = LoggerFactory.getLogger(PredictionService.class);

    private final UtilizationRepository utilizationRepository;
    private final PredictionRepository predictionRepository;
    private final PredictorRepository predictorRepository;
    private final PlatformTransactionManager transactionManager;
    private final Map<String, Predictor> predictorsByType;

    public PredictionService(UtilizationRepository utilizationRepository,
                             PredictionRepository predictionRepository,
                             PredictorRepository predictorRepository,
                             PlatformTransactionManager transactionManager,
                             Predictor... predictors) {
        this.utilizationRepository = utilizationRepository;
        this.predictionRepository = predictionRepository;
        this.predictorRepository = predictorRepository;
        this.transactionManager = transactionManager;
        Map<String, Predictor> predictorsByType = new HashMap<>();
        for (Predictor predictor : predictors) {
            predictorsByType.put(predictor.getType(), predictor);
        }
        this.predictorsByType = Collections.unmodifiableMap(predictorsByType);
    }

    @TransactionalWrite
    public void signalUpdateNeeded(List<Utilization> utilizations) {
        utilizations.stream()
                .map(Utilization::getUtilizationKey)
                .distinct()
                .forEach(this::signalUpdateNeeded);
    }

    private void signalUpdateNeeded(UtilizationKey utilizationKey) {
        // TODO: should we cache that which predictors are already enabled? it would require resetting the cache when DevHelper.deleteAll() is called
        predictorsByType.keySet().stream()
                .forEach(predictorType -> predictorRepository.enablePredictor(predictorType, utilizationKey));
        predictorRepository.markPredictorsNeedAnUpdate(utilizationKey);
    }

    private Optional<Predictor> getPredictor(String predictorType) {
        Optional<Predictor> predictor = Optional.ofNullable(predictorsByType.get(predictorType));
        if (!predictor.isPresent()) {
            log.warn("Predictor {} not installed", predictorType);
        }
        return predictor;
    }

    public Optional<PredictionBatch> getPrediction(UtilizationKey utilizationKey, DateTime time) {
        return predictionRepository.getPrediction(utilizationKey, time);
    }

    public List<PredictionBatch> getPredictionsByFacility(Long facilityId, DateTime time) {
        return predictionRepository.getPredictionsByFacility(facilityId, time);
    }

    @Scheduled(cron = "0 */5 * * * *") // every 5 minutes to match PredictionDao.PREDICTION_RESOLUTION
    public void updatePredictions() {
        log.info("updatePredictions");
        TransactionTemplate txTemplate = new TransactionTemplate(transactionManager);
        txTemplate.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED); // TODO: set in Core/JdbcConfiguration
        txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

        for (Long predictorId : findPredictorsNeedingUpdate()) {
            txTemplate.execute(tx -> {
                updatePredictor(predictorId);
                return null;
            });
        }
    }

    private List<Long> findPredictorsNeedingUpdate() {
        List<Long> predictorIds = predictorRepository.findPredictorsNeedingUpdate();
        Collections.shuffle(predictorIds); // distribute the load over all servers
        return predictorIds;
    }

    private void updatePredictor(Long predictorId) {
        PredictorState state = predictorRepository.getForUpdate(predictorId);
        state.moreUtilizations = false; // by default mark everything as processed, but allow the predictor to override it (and uninstalled predictors get disabled)
        getPredictor(state.predictorType).ifPresent(predictor -> {
            // TODO: consider the update interval of prediction types? or leave that up to the predictor?
            List<Prediction> predictions = predictor.predict(state, new UtilizationHistoryImpl(utilizationRepository, state.utilizationKey));
            // TODO: should we set state.latestUtilization here so that all predictors don't need to remember do it? or will some predictors use different logic for it, for example if they process only part of the updates?
            // TODO: save to prediction log
            predictionRepository.updatePredictions(toPredictionBatch(state, predictions), predictorId);
        });
        predictorRepository.save(state); // save state even if predictor is not present: this disables uninstalled predictors<commit_msg>fix(prediction): do not update predictor if moreUtilizations = false - this means that another node in cluster already updated the predictor.<commit_after>// Copyright © 2015 HSL <https://www.hsl.fi>
// This program is dual-licensed under the EUPL v1.2 and AGPLv3 licenses.

package fi.hsl.parkandride.core.service;

import fi.hsl.parkandride.core.back.PredictionRepository;
import fi.hsl.parkandride.core.back.PredictorRepository;
import fi.hsl.parkandride.core.back.UtilizationRepository;
import fi.hsl.parkandride.core.domain.Utilization;
import fi.hsl.parkandride.core.domain.UtilizationKey;
import fi.hsl.parkandride.core.domain.prediction.*;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.*;

import static org.springframework.transaction.annotation.Isolation.READ_COMMITTED;
import static org.springframework.transaction.annotation.Propagation.REQUIRES_NEW;

public class PredictionService {

    private static final Logger log = LoggerFactory.getLogger(PredictionService.class);

    private final UtilizationRepository utilizationRepository;
    private final PredictionRepository predictionRepository;
    private final PredictorRepository predictorRepository;
    private final PlatformTransactionManager transactionManager;
    private final Map<String, Predictor> predictorsByType;

    public PredictionService(UtilizationRepository utilizationRepository,
                             PredictionRepository predictionRepository,
                             PredictorRepository predictorRepository,
                             PlatformTransactionManager transactionManager,
                             Predictor... predictors) {
        this.utilizationRepository = utilizationRepository;
        this.predictionRepository = predictionRepository;
        this.predictorRepository = predictorRepository;
        this.transactionManager = transactionManager;
        Map<String, Predictor> predictorsByType = new HashMap<>();
        for (Predictor predictor : predictors) {
            predictorsByType.put(predictor.getType(), predictor);
        }
        this.predictorsByType = Collections.unmodifiableMap(predictorsByType);
    }

    @TransactionalWrite
    public void signalUpdateNeeded(List<Utilization> utilizations) {
        utilizations.stream()
                .map(Utilization::getUtilizationKey)
                .distinct()
                .forEach(this::signalUpdateNeeded);
    }

    private void signalUpdateNeeded(UtilizationKey utilizationKey) {
        // TODO: should we cache that which predictors are already enabled? it would require resetting the cache when DevHelper.deleteAll() is called
        predictorsByType.keySet().stream()
                .forEach(predictorType -> predictorRepository.enablePredictor(predictorType, utilizationKey));
        predictorRepository.markPredictorsNeedAnUpdate(utilizationKey);
    }

    private Optional<Predictor> getPredictor(String predictorType) {
        Optional<Predictor> predictor = Optional.ofNullable(predictorsByType.get(predictorType));
        if (!predictor.isPresent()) {
            log.warn("Predictor {} not installed", predictorType);
        }
        return predictor;
    }

    public Optional<PredictionBatch> getPrediction(UtilizationKey utilizationKey, DateTime time) {
        return predictionRepository.getPrediction(utilizationKey, time);
    }

    public List<PredictionBatch> getPredictionsByFacility(Long facilityId, DateTime time) {
        return predictionRepository.getPredictionsByFacility(facilityId, time);
    }

    @Scheduled(cron = "0 */5 * * * *") // every 5 minutes to match PredictionDao.PREDICTION_RESOLUTION
    public void updatePredictions() {
        log.info("updatePredictions");
        TransactionTemplate txTemplate = new TransactionTemplate(transactionManager);
        txTemplate.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED); // TODO: set in Core/JdbcConfiguration
        txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

        for (Long predictorId : findPredictorsNeedingUpdate()) {
            txTemplate.execute(tx -> {
                updatePredictor(predictorId);
                return null;
            });
        }
    }

    private List<Long> findPredictorsNeedingUpdate() {
        List<Long> predictorIds = predictorRepository.findPredictorsNeedingUpdate();
        Collections.shuffle(predictorIds); // distribute the load over all servers
        return predictorIds;
    }

    private void updatePredictor(Long predictorId) {
        PredictorState state = predictorRepository.getForUpdate(predictorId);
        if (state.moreUtilizations == false) {
            log.debug("Another cluster node already updated predictor ID {} (type {} for {}), skipping...", state.predictorId, state.predictorType, state.utilizationKey);
            return;
        }
        state.moreUtilizations = false; // by default mark everything as processed, but allow the predictor to override it (and uninstalled predictors get disabled)
        getPredictor(state.predictorType).ifPresent(predictor -> {
            // TODO: consider the update interval of prediction types? or leave that up to the predictor?
            List<Prediction> predictions = predictor.predict(state, new UtilizationHistoryImpl(utilizationRepository, state.utilizationKey));
            // TODO: should we set state.latestUtilization here so that all predictors don't need to remember do it? or will some predictors use different logic for it, for example if they process only part of the updates?
            // TODO: save to prediction log
            predictionRepository.updatePredictions(toPredictionBatch(state, predictions), predictorId);
        });
        predictorRepository.save(state); // save state even if predictor is not present: this disables uninstalled predictors
{'fraction_non_alphanumeric': 0.07599537419461425, 'fraction_numerical': 0.0019824880224681976, 'mean_word_length': 3.6817478731631863, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 33, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, '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': '3231665', 'n_tokens_mistral': 3017, 'n_tokens_neox': 2865, 'n_words': 712}