<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.back.prediction;

import com.querydsl.core.Tuple;
import com.querydsl.sql.dml.SQLInsertClause;
import com.querydsl.sql.dml.SQLUpdateClause;
import com.querydsl.sql.postgresql.PostgreSQLQueryFactory;
import com.querydsl.core.types.Expression;
import com.querydsl.core.types.MappingProjection;
import com.querydsl.core.types.Path;
import com.querydsl.core.types.Projections;
import com.querydsl.core.types.dsl.BooleanExpression;
import fi.hsl.parkandride.back.TimeUtil;
import fi.hsl.parkandride.back.sql.QFacilityPrediction;
import fi.hsl.parkandride.back.sql.QFacilityPredictionHistory;
import fi.hsl.parkandride.core.back.PredictionRepository;
import fi.hsl.parkandride.core.domain.UtilizationKey;
import fi.hsl.parkandride.core.domain.prediction.Prediction;
import fi.hsl.parkandride.core.domain.prediction.PredictionBatch;
import fi.hsl.parkandride.core.service.TransactionalRead;
import fi.hsl.parkandride.core.service.TransactionalWrite;
import fi.hsl.parkandride.core.service.ValidationService;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Duration;
import org.joda.time.format.DateTimeFormat;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.stream.Collectors.toList;
import static org.joda.time.Duration.standardHours;
import static org.joda.time.Duration.standardMinutes;
import static org.springframework.transaction.annotation.Isolation.READ_COMMITTED;
import static org.springframework.transaction.annotation.Isolation.SERIALIZABLE;
import static org.springframework.transaction.annotation.Propagation.REQUIRED;

public class PredictionDao implements PredictionRepository {

    private static final QFacilityPrediction qPrediction = QFacilityPrediction.facilityPrediction;
    private static final QFacilityPredictionHistory qPredictionHistory = QFacilityPredictionHistory.facilityPredictionHistory;
    private static final Map<String, Path<Integer>> spacesAvailableColumnsByHHmm = Collections.unmodifiableMap(
            Stream.of(qPrediction.all())
                    .filter(p -> p.getMetadata().getName().startsWith("spacesAvailableAt"))
                    .map(PredictionDao::castToIntegerPath)
                    .collect(Collectors.toMap(
                            p -> p.getMetadata().getName().substring("spacesAvailableAt".length()),
                            Function.identity())));

    private final PostgreSQLQueryFactory queryFactory;
    private final ValidationService validationService;
    private final List<Duration> predictionsDistancesToStore = Collections.unmodifiableList(
            Arrays.<Duration>asList(standardMinutes(5), standardMinutes(10), standardMinutes(15), standardMinutes(20),
                    standardMinutes(30), standardMinutes(45), standardHours(1), standardHours(2), standardHours(4),
                    standardHours(8), standardHours(12), standardHours(16), standardHours(20), standardHours(24)));

    public PredictionDao(PostgreSQLQueryFactory queryFactory, ValidationService validationService) {
        this.queryFactory = queryFactory;
        this.validationService = validationService;
    }

    @Transactional(readOnly = false, isolation = SERIALIZABLE, propagation = REQUIRED)
    @Override
    public void updatePredictions(PredictionBatch pb, Long predictorId) {
        validationService.validate(pb);
        UtilizationKey utilizationKey = pb.utilizationKey;
        DateTime start = toPredictionResolution(pb.sourceTimestamp);
        List<Prediction> predictions = normalizeToPredictionWindow(start, pb.predictions);

        long updatedRows = maybeUpdatePredictionLookupTable(utilizationKey, start, predictions);
        if (updatedRows == 0) {
            initializePredictionLookupTable(utilizationKey);
            updatePredictions(pb, predictorId); // retry now that the lookup table exists
        } else {
            savePredictionHistory(predictorId, start, predictions);
        }
    }

    @TransactionalWrite
    @Override
    public void updateOnlyPredictionHistory(PredictionBatch pb, Long predictorId) {
        validationService.validate(pb);
        DateTime start = toPredictionResolution(pb.sourceTimestamp);
        List<Prediction> predictions = normalizeToPredictionWindow(start, pb.predictions);
        predictions = predictions.stream()
                .filter(p -> predictionsDistancesToStore.contains(new Duration(start, p.timestamp)))
                .collect(toList());
        savePredictionHistory(predictorId, start, predictions);
    }

    private static List<Prediction> normalizeToPredictionWindow(DateTime start, List<Prediction> predictions) {
        DateTime end = start.plus(PREDICTION_WINDOW).minus(PREDICTION_RESOLUTION);
        return predictions.stream()
                // remove too fine-grained predictions
                .collect(groupByRoundedTimeKeepingNewest()) // -> Map<DateTime, Prediction>
                .values().stream()
                        // normalize resolution
                .map(roundTimestampsToPredictionResolution())
                        // interpolate too coarse-grained predictions
                .sorted(Comparator.comparing(p -> p.timestamp))
                .map(Collections::singletonList)                            // 1. wrap values in immutable singleton lists
                .reduce(new ArrayList<>(), linearInterpolation()).stream()  // 2. mutable ArrayList as accumulator
                        // normalize range
                .filter(isWithin(start, end)) // after interpolation because of PredictionDaoTest.does_linear_interpolation_also_between_values_outside_the_prediction_window
                .collect(toList());
    }
<commit_msg>refactor(prediction): separate prediction filtering to selected distances into a method for later reuse.<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.back.prediction;

import com.querydsl.core.Tuple;
import com.querydsl.sql.dml.SQLInsertClause;
import com.querydsl.sql.dml.SQLUpdateClause;
import com.querydsl.sql.postgresql.PostgreSQLQueryFactory;
import com.querydsl.core.types.Expression;
import com.querydsl.core.types.MappingProjection;
import com.querydsl.core.types.Path;
import com.querydsl.core.types.Projections;
import com.querydsl.core.types.dsl.BooleanExpression;
import fi.hsl.parkandride.back.TimeUtil;
import fi.hsl.parkandride.back.sql.QFacilityPrediction;
import fi.hsl.parkandride.back.sql.QFacilityPredictionHistory;
import fi.hsl.parkandride.core.back.PredictionRepository;
import fi.hsl.parkandride.core.domain.UtilizationKey;
import fi.hsl.parkandride.core.domain.prediction.Prediction;
import fi.hsl.parkandride.core.domain.prediction.PredictionBatch;
import fi.hsl.parkandride.core.service.TransactionalRead;
import fi.hsl.parkandride.core.service.TransactionalWrite;
import fi.hsl.parkandride.core.service.ValidationService;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Duration;
import org.joda.time.format.DateTimeFormat;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.stream.Collectors.toList;
import static org.joda.time.Duration.standardHours;
import static org.joda.time.Duration.standardMinutes;
import static org.springframework.transaction.annotation.Isolation.READ_COMMITTED;
import static org.springframework.transaction.annotation.Isolation.SERIALIZABLE;
import static org.springframework.transaction.annotation.Propagation.REQUIRED;

public class PredictionDao implements PredictionRepository {

    private static final QFacilityPrediction qPrediction = QFacilityPrediction.facilityPrediction;
    private static final QFacilityPredictionHistory qPredictionHistory = QFacilityPredictionHistory.facilityPredictionHistory;
    private static final Map<String, Path<Integer>> spacesAvailableColumnsByHHmm = Collections.unmodifiableMap(
            Stream.of(qPrediction.all())
                    .filter(p -> p.getMetadata().getName().startsWith("spacesAvailableAt"))
                    .map(PredictionDao::castToIntegerPath)
                    .collect(Collectors.toMap(
                            p -> p.getMetadata().getName().substring("spacesAvailableAt".length()),
                            Function.identity())));

    private final PostgreSQLQueryFactory queryFactory;
    private final ValidationService validationService;
    private final List<Duration> predictionsDistancesToStore = Collections.unmodifiableList(
            Arrays.<Duration>asList(standardMinutes(5), standardMinutes(10), standardMinutes(15), standardMinutes(20),
                    standardMinutes(30), standardMinutes(45), standardHours(1), standardHours(2), standardHours(4),
                    standardHours(8), standardHours(12), standardHours(16), standardHours(20), standardHours(24)));

    public PredictionDao(PostgreSQLQueryFactory queryFactory, ValidationService validationService) {
        this.queryFactory = queryFactory;
        this.validationService = validationService;
    }

    @Transactional(readOnly = false, isolation = SERIALIZABLE, propagation = REQUIRED)
    @Override
    public void updatePredictions(PredictionBatch pb, Long predictorId) {
        validationService.validate(pb);
        UtilizationKey utilizationKey = pb.utilizationKey;
        DateTime start = toPredictionResolution(pb.sourceTimestamp);
        List<Prediction> predictions = normalizeToPredictionWindow(start, pb.predictions);

        long updatedRows = maybeUpdatePredictionLookupTable(utilizationKey, start, predictions);
        if (updatedRows == 0) {
            initializePredictionLookupTable(utilizationKey);
            updatePredictions(pb, predictorId); // retry now that the lookup table exists
        } else {
            savePredictionHistory(predictorId, start, predictions);
        }
    }

    @TransactionalWrite
    @Override
    public void updateOnlyPredictionHistory(PredictionBatch pb, Long predictorId) {
        validationService.validate(pb);
        DateTime start = toPredictionResolution(pb.sourceTimestamp);
        List<Prediction> predictions = normalizeToPredictionWindow(start, pb.predictions);
        savePredictionHistory(predictorId, start, filterToSelectedPredictionDistances(start, predictions));
    }

    private List<Prediction> filterToSelectedPredictionDistances(DateTime start, List<Prediction> predictions) {
        return predictions.stream()
                .filter(p -> predictionsDistancesToStore.contains(new Duration(start, p.timestamp)))
                .collect(toList());
    }

    private static List<Prediction> normalizeToPredictionWindow(DateTime start, List<Prediction> predictions) {
        DateTime end = start.plus(PREDICTION_WINDOW).minus(PREDICTION_RESOLUTION);
        return predictions.stream()
                // remove too fine-grained predictions
                .collect(groupByRoundedTimeKeepingNewest()) // -> Map<DateTime, Prediction>
                .values().stream()
                        // normalize resolution
                .map(roundTimestampsToPredictionResolution())
                        // interpolate too coarse-grained predictions
                .sorted(Comparator.comparing(p -> p.timestamp))
                .map(Collections::singletonList)                            // 1. wrap values in immutable singleton lists
                .reduce(new ArrayList<>(), linearInterpolation()).stream()  // 2. mutable ArrayList as accumulator
                        // normalize range
                .filter(isWithin(start, end)) // after interpolation because of PredictionDaoTest.does_linear_interpolation_also_between_values_outside_the_prediction_window
                .collect(toList());
    }

{'fraction_non_alphanumeric': 0.09129528695094581, 'fraction_numerical': 0.00529015710163514, 'mean_word_length': 3.7640320733104238, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 37, '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': '7153687', 'n_tokens_mistral': 3374, 'n_tokens_neox': 3275, 'n_words': 607}