81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
# meteo/gaps.py
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import List
|
|
|
|
import pandas as pd
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TimeGap:
|
|
"""
|
|
Représente une période pendant laquelle il manque des points dans une
|
|
série temporelle censée être échantillonnée à intervalle régulier.
|
|
"""
|
|
|
|
# Timestamp du dernier point avant le trou
|
|
before: pd.Timestamp
|
|
# Timestamp du premier point après le trou
|
|
after: pd.Timestamp
|
|
|
|
# Premier timestamp "attendu" manquant
|
|
missing_start: pd.Timestamp
|
|
# Dernier timestamp "attendu" manquant
|
|
missing_end: pd.Timestamp
|
|
|
|
# Nombre d'intervalles manquants (par ex. 3 => 3 minutes manquantes)
|
|
missing_intervals: int
|
|
|
|
# Durée totale du gap (after - before)
|
|
duration: pd.Timedelta
|
|
|
|
|
|
def find_time_gaps(
|
|
df: pd.DataFrame,
|
|
expected_freq: pd.Timedelta = pd.Timedelta(minutes=1),
|
|
) -> List[TimeGap]:
|
|
"""
|
|
Détecte les gaps temporels dans un DataFrame indexé par le temps.
|
|
|
|
Un "gap" est un intervalle entre deux timestamps successifs strictement
|
|
supérieur à `expected_freq`.
|
|
|
|
Exemple : si expected_freq = 1 minute et qu'on passe de 10:00 à 10:05,
|
|
on détecte un gap avec 4 minutes manquantes (10:01, 10:02, 10:03, 10:04).
|
|
"""
|
|
if not isinstance(df.index, pd.DatetimeIndex):
|
|
raise TypeError(
|
|
"find_time_gaps nécessite un DataFrame avec un DatetimeIndex."
|
|
)
|
|
|
|
index = df.index.sort_values()
|
|
diffs = index.to_series().diff()
|
|
|
|
gaps: list[TimeGap] = []
|
|
|
|
for i, delta in enumerate(diffs.iloc[1:], start=1):
|
|
if delta <= expected_freq:
|
|
continue
|
|
|
|
before_ts = index[i - 1]
|
|
after_ts = index[i]
|
|
|
|
# Nombre d'intervalles manquants (ex: 5min / 1min => 4 intervalles manquants)
|
|
missing_intervals = int(delta // expected_freq) - 1
|
|
|
|
missing_start = before_ts + expected_freq
|
|
missing_end = after_ts - expected_freq
|
|
|
|
gap = TimeGap(
|
|
before=before_ts,
|
|
after=after_ts,
|
|
missing_start=missing_start,
|
|
missing_end=missing_end,
|
|
missing_intervals=missing_intervals,
|
|
duration=delta,
|
|
)
|
|
gaps.append(gap)
|
|
|
|
return gaps
|