vendimia-compose/backend/routes/calculations/pipelines.py

from config.settings import DECIMAL_PLACES


def truncate_100(num):
    return 100 if num > 100 else num


def get_percent(count, total):
    return truncate_100(round((count * 100) / total, DECIMAL_PLACES))


id_daily = {
    "day": {"$dayOfMonth": "$date"},
    "month": {"$month": "$date"},
    "year": {"$year": "$date"},
}


def ltgt_pipeline(start_datetime, end_datetime):
    return [
        {
            "$match": {
                "date": {
                    "$gte": start_datetime,
                    "$lte": end_datetime,
                },
                "temp": {
                    "$ne": None,
                },
            },
        },
        {
            "$group": {
                "_id": None,
                "lt10_count": {
                    "$sum": {"$cond": [{"$lt": ["$temp", 10]}, 1, 0]},
                },
                "gt30_count": {
                    "$sum": {"$cond": [{"$gt": ["$temp", 30]}, 1, 0]},
                },
                "gt33_count": {
                    "$sum": {"$cond": [{"$gt": ["$temp", 33]}, 1, 0]},
                },
                "count": {
                    "$sum": 1,
                },
            },
        },
    ]


def deg_daily_acc_avg_pipeline(start_datetime, end_datetime):
    return [
        {
            "$match": {  # Conseguimos las muestras
                "date": {
                    "$gte": start_datetime,
                    "$lte": end_datetime,
                },
                "temp": {
                    "$ne": None,
                },
            },
        },
        {
            "$group": {  # Agrupamos por dia y calculamos el promedio de temperatura(de ese dia)
                "_id": id_daily,
                "avg": {
                    "$avg": "$temp",
                },
            }
        },
        {
            "$addFields": {  # Pisamos el campo promedio que calculamos antes y lo usamos para los grados dia de ese dia(medio feo)
                "avg": {
                    "$cond": [
                        {"$gt": ["$avg", 10]},
                        {"$add": ["$avg", -10]  # Cantidad de grados por encima de la temperatura base
                         },
                        0,
                    ],
                },
            },
        },
        {
            "$group": {
                "_id": None,
                "sum": {  # Grados dias acumulados
                    "$sum": "$avg",
                },
                "avg": {  # grados dias promedio
                    "$avg": "$avg",
                },
            },
        },
    ]


def daily_thermal_amplitude_pipeline(start_datetime, end_datetime):
    return [
        {
            "$match": {
                "date": {
                    "$gte": start_datetime,
                    "$lte": end_datetime,
                },
                "temp": {"$ne": None},
            },
        },
        {
            "$group": {
                "_id": id_daily,
                "min": {
                    "$min": "$temp",
                },
                "max": {
                    "$max": "$temp",
                },
            }
        },
        {
            "$addFields": {
                "amplitude": {
                    "$subtract": ["$max", "$min"],
                },
            },
        },
        {
            "$group": {
                "_id": None,
                "avg": {
                    "$avg": "$amplitude",
                },
            },
        },
    ]


def precip_acc_pipeline(start_datetime, end_datetime):
    return [
        {
            "$match": {
                "date": {
                    "$gte": start_datetime,
                    "$lte": end_datetime,
                },
                "precip": {"$ne": None},
            },
        },
        {
            "$group": {
                "_id": None,
                "sum": {
                    "$sum": "$precip",
                },
            },
        },
    ]


def count_measures_pipeline(start_datetime, end_datetime) -> dict:
    return {
        "date": {
            "$gte": start_datetime,
            "$lte": end_datetime,
        },
        "$or": [
            {
                "temp": {
                    "$ne": None,
                },
            },
            {
                "precip": {
                    "$ne": None,
                },
            },
        ],
    }


def count_temp_measures_pipeline(start_datetime, end_datetime) -> dict:
    return {
        "date": {
            "$gte": start_datetime,
            "$lte": end_datetime,
        },
        "temp": {
            "$ne": None,
        },
    }


def count_precip_measures_pipeline(start_datetime, end_datetime) -> dict:
    return {
        "date": {
            "$gte": start_datetime,
            "$lte": end_datetime,
        },
        "precip": {
            "$ne": None,
        },
    }