agreg-server/server/backend/insert_dummy_data.py

"""
Pour ajouter les données et exécuter le script, créer le conteneur :
docker run --rm --network server_database simugaz/backend:latest insert_dummy_data.py <options>
"""
import argparse
from datetime import datetime, timedelta
from random import random
from typing import NamedTuple
from zoneinfo import ZoneInfo

import psycopg2

TZ = ZoneInfo("Europe/Paris")
INITIAL_CONSUMPTION = 1854.32
AVERAGE_DAILY_CONSUMPTION = 2.66
PULSE_RESOLUTION = 0.010
DB_URI = "postgresql://simugaz:simugaz@db/simugaz"

GRANULARITIES = {
    "hour":  3600,
    "day":   86400
}

GRDF_MONTHLY_COEFF = [
    2.21,  # Janvier
    2.05,  # Février
    1.62,  # Mars
    0.97,  # Avril
    0.55,  # Mai
    0.22,  # Juin
    0.15,  # Juillet
    0.17,  # Août
    0.38,  # Septembre
    0.85,  # Octobre
    1.52,  # Novembre
    2.11,  # Décembre
]


class PulseReading(NamedTuple):
    device_id: str
    date: datetime
    value: int


def simulate_gas_pulses(
    device_id: str,
    initial_consumption: float,
    start_date: datetime,
    end_date: datetime,
    granularity_seconds: int = 3600,
) -> list[PulseReading]:
    results: list[PulseReading] = []
    consumption_m3 = initial_consumption
    current_date = start_date

    while current_date < end_date:
        daily_base = (
            AVERAGE_DAILY_CONSUMPTION * GRDF_MONTHLY_COEFF[current_date.month - 1]
        )
        daily_variation = 1.0 + 0.30 * (random() - 0.5)

        increment_m3 = daily_base * granularity_seconds / 86400.0 * daily_variation
        consumption_m3 += increment_m3

        pulses = int(consumption_m3 / PULSE_RESOLUTION)
        new_reading = PulseReading(device_id=device_id, date=current_date, value=pulses)
        results.append(new_reading)

        current_date += timedelta(seconds=granularity_seconds)

    return results


def write_to_db(points: list[PulseReading]) -> None:
    with psycopg2.connect(DB_URI) as conn:
        with conn.cursor() as cur:
            cur.executemany(
                """
                INSERT INTO reading (device_id, date, pulses)
                VALUES (%s, %s, %s)
                ON CONFLICT DO NOTHING
                """,
                points,
            )
        conn.commit()
    print(f"[DB] {len(points)} lignes insérées")


def parse_args():
    parser = argparse.ArgumentParser(description="Générateur de données SimuGaz")

    parser.add_argument(
        "--start",
        type=lambda s: datetime.fromisoformat(s).replace(tzinfo=TZ),
        default=datetime(2025, 1, 1, 0, 0, 0, tzinfo=TZ),
        help="Date de début (défaut : 2025-01-01)",
    )
    parser.add_argument(
        "--end",
        type=lambda s: datetime.fromisoformat(s).replace(tzinfo=TZ),
        default=datetime.now(tz=TZ),
        help="Date de fin (défaut : maintenant)",
    )
    parser.add_argument(
        "--device-id",
        type=str,
        default=None,
        metavar="UUID",
        help="UUID du device",
    )
    parser.add_argument(
        "--conso",
        type=float,
        default=INITIAL_CONSUMPTION,
        help=f"Compteur de consommation initial (défaut : {INITIAL_CONSUMPTION})",
    )
    parser.add_argument(
        "--granularity",
        choices=GRANULARITIES.keys(),
        default="hour",
        help="Granularité des points ('hour' par défaut)",
    )
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()

    points = simulate_gas_pulses(
        device_id=args.device_id,
        initial_consumption=args.conso,
        start_date=args.start,
        end_date=args.end,
        granularity_seconds=GRANULARITIES[args.granularity],
    )

    write_to_db(points)