تنظيف البيانات ومعالجتها

raw = pd.read_csv("datasets/node_log_raw.csv", parse_dates=["timestamp"])
print("rows:", len(raw), "| duplicates:", raw.duplicated(["timestamp", "node_id"]).sum())
df = raw.drop_duplicates(["timestamp", "node_id"]).copy()

limits = {"air_temp_c": (-10, 60), "air_humidity_pct": (0, 100), "soil_moisture_pct": (0, 100)}
for col, (lo, hi) in limits.items():
    bad = ~df[col].between(lo, hi) & df[col].notna()
    print(f"{col}: {bad.sum()} impossible values -> NaN")
    df.loc[bad, col] = np.nan

g = df[df.node_id == "field_B"].set_index("timestamp").sort_index()
grid = pd.date_range(g.index.min(), g.index.max(), freq="15min")
g = g.reindex(grid)                                  # the 7-hour outage now appears as NaN rows
print("missing rows after reindex:", g["air_temp_c"].isna().sum())

def hampel(s, window=9, k=3.0):
    med = s.rolling(window, center=True, min_periods=1).median()
    mad = (s - med).abs().rolling(window, center=True, min_periods=1).median()
    return (s - med).abs() > k * 1.4826 * mad.replace(0, np.nan)

spikes = hampel(g["air_temp_c"])
print("Hampel spikes in air_temp_c:", int(spikes.sum()))
g.loc[spikes, "air_temp_c"] = np.nan
g["air_temp_c"] = g["air_temp_c"].interpolate(method="time", limit=8)   # fill gaps up to 2 h only
print("still missing (long outage kept on purpose):", g["air_temp_c"].isna().sum())

h = g[["air_temp_c", "soil_moisture_pct"]].resample("h").mean()         # 15-min -> hourly
h["soil_norm"] = (h.soil_moisture_pct - h.soil_moisture_pct.min()) / (h.soil_moisture_pct.max() - h.soil_moisture_pct.min())
h["temp_z"] = (h.air_temp_c - h.air_temp_c.mean()) / h.air_temp_c.std()
h["temp_roll_6h"] = h.air_temp_c.rolling(6, min_periods=3).mean()       # smoothed trend
h["soil_lag_24h"] = h.soil_moisture_pct.shift(24)                        # same hour yesterday
h["soil_change_24h"] = h.soil_moisture_pct - h.soil_lag_24h              # drying rate per day
print(h.dropna().head(3).round(3))
print("mean daily drying rate (%/day):", round(h.soil_change_24h.mean(), 2))

"""Session 14 - Data cleaning & preprocessing of the farm nodes log.

Input : datasets/node_log_raw.csv  (3 nodes, 15-min, with real-world faults:
        missing values, -999/999 sensor errors, a 7-hour Wi-Fi outage, duplicated MQTT messages)
Output: outputs/node_log_clean.csv  + a cleaning report printed to the console
"""
import numpy as np
import pandas as pd

from common import DATA, OUT

raw = pd.read_csv(DATA / "node_log_raw.csv", parse_dates=["timestamp"])
print("Raw rows:", len(raw))
report = {}

# 1) Duplicates (MQTT QoS 1 can deliver the same message twice)
report["duplicates"] = int(raw.duplicated(["timestamp", "node_id"]).sum())
df = raw.drop_duplicates(["timestamp", "node_id"]).copy()

# 2) Physically impossible values -> NaN (DHT22 returns -999 / 999 on read errors)
limits = {"air_temp_c": (-10, 60), "air_humidity_pct": (0, 100), "soil_moisture_pct": (0, 100)}
for col, (lo, hi) in limits.items():
    bad = ~df[col].between(lo, hi) & df[col].notna()
    report[f"out_of_range_{col}"] = int(bad.sum())
    df.loc[bad, col] = np.nan

# 3) Statistical outliers with a rolling median + MAD (robust to spikes)
def hampel(s: pd.Series, window: int = 9, k: float = 3.0) -> pd.Series:
    med = s.rolling(window, center=True, min_periods=1).median()
    mad = (s - med).abs().rolling(window, center=True, min_periods=1).median()
    return (s - med).abs() > k * 1.4826 * mad.replace(0, np.nan)

clean_parts = []
for node, g in df.groupby("node_id"):
    g = g.set_index("timestamp").sort_index()
    # 4) Regular time grid: reveals the gaps (outage) as NaN rows
    g = g.reindex(pd.date_range(g.index.min(), g.index.max(), freq="15min"))
    g["node_id"] = node
    report[f"missing_rows_{node}"] = int(g["air_temp_c"].isna().sum())
    for col in ["air_temp_c", "air_humidity_pct", "soil_moisture_pct"]:
        out = hampel(g[col])
        report[f"hampel_{node}_{col}"] = int(out.sum())
        g.loc[out, col] = np.nan
        # 5) Temporal interpolation - only for short gaps (<= 2 hours = 8 samples)
        g[col] = g[col].interpolate(method="time", limit=8)
    g["still_missing"] = g["air_temp_c"].isna()         # still NaN after the 2-hour limit -> real outage
    clean_parts.append(g)

clean = pd.concat(clean_parts).rename_axis("timestamp").reset_index()

# 6) Normalisation (min-max per variable) - ready for ML
for col in ["air_temp_c", "air_humidity_pct", "soil_moisture_pct"]:
    lo, hi = clean[col].min(), clean[col].max()
    clean[f"{col}_norm"] = ((clean[col] - lo) / (hi - lo)).round(4)

# 7) Time-series transformation: hourly means per node + daily features
hourly = (clean.set_index("timestamp").groupby("node_id")
          .resample("h")[["air_temp_c", "air_humidity_pct", "soil_moisture_pct"]].mean().round(2))
clean.to_csv(OUT / "node_log_clean.csv", index=False)
hourly.to_csv(OUT / "node_log_hourly.csv")

print("\n== Cleaning report ==")
for k, v in report.items():
    print(f"{k:45s} {v}")
print("\nRemaining NaN (long outages kept as NaN on purpose):")
print(clean[["air_temp_c", "air_humidity_pct", "soil_moisture_pct"]].isna().sum())
print("\nSaved:", OUT / "node_log_clean.csv")