generate_data.py

"""
Battery Sensor Data Generator
Generates synthetic battery monitoring data with temporal cycling patterns
for BMS AI model training, testing, and visualization.
"""

import numpy as np
import pandas as pd
from datetime import datetime, timedelta


def classify_condition(voltage, current, temperature):
    """Classify battery condition based on sensor thresholds."""
    # Unsafe conditions
    if voltage < 2.8 or voltage > 4.4:
        return "Unsafe"
    if temperature > 55:
        return "Unsafe"
    if current > 4.0:
        return "Unsafe"
    if voltage < 2.8 and temperature > 50:
        return "Unsafe"

    # Warning conditions
    if voltage < 3.2 or voltage > 4.25:
        return "Warning"
    if temperature > 42 or temperature < 5:
        return "Warning"
    if current > 2.8:
        return "Warning"

    # Normal
    return "Normal"


def classify_application(voltage, current, temperature, soc, cycle_count):
    """
    Classify battery application suitability based on composite health score.

    Returns one of:
        - High-Performance: EVs, power tools, drones
        - Moderate-Load: laptops, home appliances
        - Low-Power: IoT sensors, remote controls, LED lights
        - Unsuitable: recycling/disposal
    """
    # --- Compute health score (0–100) ---
    # Voltage score (optimal: 3.6–4.1V)
    if 3.6 <= voltage <= 4.1:
        v_score = 100
    elif 3.3 <= voltage < 3.6 or 4.1 < voltage <= 4.25:
        v_score = 65
    elif 3.0 <= voltage < 3.3 or 4.25 < voltage <= 4.4:
        v_score = 35
    else:
        v_score = 0

    # Temperature score (optimal: 20–35°C)
    if 20 <= temperature <= 35:
        t_score = 100
    elif 10 <= temperature < 20 or 35 < temperature <= 42:
        t_score = 65
    elif 5 <= temperature < 10 or 42 < temperature <= 55:
        t_score = 30
    else:
        t_score = 0

    # SoC score
    if soc >= 70:
        s_score = 100
    elif soc >= 40:
        s_score = 65
    elif soc >= 15:
        s_score = 35
    else:
        s_score = 10

    # Cycle degradation score (lower cycles = better)
    if cycle_count <= 200:
        c_score = 100
    elif cycle_count <= 500:
        c_score = 70
    elif cycle_count <= 800:
        c_score = 40
    else:
        c_score = 15

    # Current draw score (lower = more versatile)
    if current <= 1.5:
        i_score = 100
    elif current <= 2.5:
        i_score = 70
    elif current <= 3.5:
        i_score = 35
    else:
        i_score = 5

    # Weighted composite score
    health_score = (
        0.25 * v_score +
        0.20 * t_score +
        0.25 * s_score +
        0.15 * c_score +
        0.15 * i_score
    )

    # Classify based on score
    if health_score >= 75:
        return "High-Performance"
    elif health_score >= 50:
        return "Moderate-Load"
    elif health_score >= 25:
        return "Low-Power"
    else:
        return "Unsuitable"


def generate_battery_data(num_samples=5000, seed=42):
    """
    Generate synthetic battery sensor data with realistic temporal patterns.

    Simulates charge/discharge cycles with gradual transitions,
    occasional anomalies, and realistic noise.
    """
    np.random.seed(seed)

    # --- Time axis: ~7 days at 2-minute intervals ---
    start_time = datetime(2026, 2, 17, 0, 0, 0)
    timestamps = [start_time + timedelta(minutes=2 * i) for i in range(num_samples)]

    # --- Simulate charge/discharge cycling ---
    # A full cycle is ~240 samples (~8 hours)
    cycle_period = 240
    phase = np.linspace(0, num_samples / cycle_period * 2 * np.pi, num_samples)

    # Base voltage follows a sinusoidal charge/discharge curve (3.2V – 4.2V)
    base_voltage = 3.7 + 0.5 * np.sin(phase)

    # Current is positive during discharge, negative during charge
    base_current = 1.5 * np.cos(phase)

    # Temperature correlates with |current| (higher current = more heat)
    base_temp = 28 + 8 * np.abs(np.sin(phase))

    # SoC inversely tracks discharge (follows voltage loosely)
    base_soc = 50 + 45 * np.sin(phase)

    # --- Inject condition-based anomalies ---
    condition_weights = np.random.choice(
        ["normal", "warning", "unsafe"],
        size=num_samples,
        p=[0.70, 0.20, 0.10],
    )

    voltages = []
    currents = []
    temperatures = []
    socs = []
    cycle_counts = []

    for i, cond in enumerate(condition_weights):
        v = base_voltage[i]
        c = abs(base_current[i])
        t = base_temp[i]
        s = np.clip(base_soc[i], 0, 100)

        if cond == "warning":
            # Push values toward warning thresholds
            v += np.random.choice([-0.6, 0.4])
            c += np.random.uniform(0.8, 1.8)
            t += np.random.uniform(8, 18)
            s = np.clip(s - np.random.uniform(15, 30), 0, 100)
        elif cond == "unsafe":
            # Push values into unsafe territory
            v += np.random.choice([-1.0, 0.8])
            c += np.random.uniform(2.0, 3.5)
            t += np.random.uniform(20, 40)
            s = np.clip(s - np.random.uniform(30, 50), 0, 100)

        # Add realistic sensor noise
        v += np.random.normal(0, 0.02)
        c = max(0, c + np.random.normal(0, 0.05))
        t += np.random.normal(0, 0.5)

        voltages.append(round(v, 3))
        currents.append(round(c, 3))
        temperatures.append(round(t, 2))
        socs.append(round(float(s), 1))
        cycle_counts.append(int(50 + i * 0.2 + np.random.randint(0, 5)))

    # --- Compute voltage rate of change (rolling delta) ---
    voltage_series = pd.Series(voltages)
    voltage_delta = voltage_series.diff().rolling(window=5, min_periods=1).mean()
    voltage_delta = voltage_delta.fillna(0).round(4).tolist()

    # --- Classify conditions based on actual generated values ---
    conditions = [
        classify_condition(v, c, t)
        for v, c, t in zip(voltages, currents, temperatures)
    ]

    # --- Classify application suitability ---
    applications = [
        classify_application(v, c, t, s, cc)
        for v, c, t, s, cc in zip(voltages, currents, temperatures, socs, cycle_counts)
    ]

    df = pd.DataFrame({
        "timestamp": timestamps,
        "voltage": voltages,
        "current": currents,
        "temperature": temperatures,
        "soc": socs,
        "cycle_count": cycle_counts,
        "voltage_delta": voltage_delta,
        "condition": conditions,
        "application": applications,
    })

    return df


if __name__ == "__main__":
    print("🔋 Generating synthetic battery data...")
    df = generate_battery_data(num_samples=5000)

    # Print distribution
    print(f"\n📊 Dataset shape: {df.shape}")
    print(f"\n📈 Condition distribution:")
    print(df["condition"].value_counts().to_string())
    print(f"\n🎯 Application suitability distribution:")
    print(df["application"].value_counts().to_string())
    print(f"\n📋 Sample data:")
    print(df.head(10).to_string(index=False))

    # Save
    output_path = "battery_data.csv"
    df.to_csv(output_path, index=False)
    print(f"\n✅ Saved to {output_path}")