98 lines
2.5 KiB
Python
98 lines
2.5 KiB
Python
|
"""
|
||
|
|
Pour ajouter les données et exécuter le script, créer le conteneur :
|
||
|
|
docker run --rm --network server_database simugaz/backend:latest insert_data.py
|
||
|
|
"""
|
||
|
|
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from random import random
|
||
|
|
from typing import NamedTuple
|
||
|
|
from zoneinfo import ZoneInfo
|
||
|
|
|
||
|
|
import psycopg2
|
||
|
|
|
||
|
|
INITIAL_CONSUMPTION = 1854.32
|
||
|
|
AVERAGE_DAILY_CONSUMPTION = 2.66
|
||
|
|
PULSE_RESOLUTION = 0.010
|
||
|
|
DB_URI = "postgresql://simugaz:simugaz@db/simugaz"
|
||
|
|
|
||
|
|
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")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
TZ = ZoneInfo("Europe/Paris")
|
||
|
|
start = datetime(2025, 1, 1, 0, 0, 0, tzinfo=TZ)
|
||
|
|
end = datetime.now(tz=TZ)
|
||
|
|
|
||
|
|
points = simulate_gas_pulses(
|
||
|
|
device_id="60124673-1999-414d-a3c5-723bd6d82a13",
|
||
|
|
initial_consumption=INITIAL_CONSUMPTION,
|
||
|
|
start_date=start,
|
||
|
|
end_date=end,
|
||
|
|
granularity_seconds=3600,
|
||
|
|
)
|
||
|
|
|
||
|
|
write_to_db(points)
|