Spaces:
Runtime error
Runtime error
File size: 9,278 Bytes
dc610d2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 | """
Seed demo data for ZenVolt platform.
Creates realistic assets based on NASA battery data.
"""
import pandas as pd
import json
from datetime import datetime, timedelta
from database import SessionLocal, Asset, HealthPrediction, init_db
from predict_soh import predict_soh
import uuid
def seed_demo_assets():
"""Create demo assets from NASA battery data."""
init_db()
db = SessionLocal()
# Clear existing data
db.query(HealthPrediction).delete()
db.query(Asset).delete()
db.commit()
print("🌱 Seeding demo data...")
# Demo user IDs
demo_users = ["demo-user-1", "demo-user-2"]
# NASA battery configurations with realistic metadata
nasa_configs = [
{
"battery": "B0005",
"project_name": "NASA B0005 - 18650 Li-ion Cell",
"asset_id": "NASA_B0005",
"location": "NASA Ames Research Center, Moffett Field, CA",
"bms_provider": "NASA Battery Testing Station (Arbin BT2000)",
"category": ["Research", "18650", "Li-ion", "Validated"],
"stage": "Certified",
"owner": demo_users[0],
"specs": {
"chemistry": "LiCoO2/Graphite",
"format": "18650 cylindrical",
"rated_capacity": "1.85 Ah",
"nominal_voltage": "3.7V",
"test_period": "2008",
"cycles_tested": "168"
}
},
{
"battery": "B0006",
"asset_id": "NASA_B0006",
"project_name": "NASA B0006 - 18650 Li-ion Cell",
"location": "NASA Ames Research Center, Moffett Field, CA",
"bms_provider": "NASA Battery Testing Station (Arbin BT2000)",
"category": ["Research", "18650", "Li-ion", "Validated"],
"stage": "Certified",
"owner": demo_users[0],
"specs": {
"chemistry": "LiCoO2/Graphite",
"format": "18650 cylindrical",
"rated_capacity": "2.04 Ah",
"nominal_voltage": "3.7V",
"test_period": "2008",
"cycles_tested": "168"
}
},
{
"battery": "B0007",
"asset_id": "NASA_B0007",
"project_name": "NASA B0007 - 18650 Li-ion Cell (Validation)",
"location": "NASA Ames Research Center, Moffett Field, CA",
"bms_provider": "NASA Battery Testing Station (Arbin BT2000)",
"category": ["Research", "18650", "Li-ion", "Validated", "Test-Set"],
"stage": "Certified",
"owner": demo_users[0],
"specs": {
"chemistry": "LiCoO2/Graphite",
"format": "18650 cylindrical",
"rated_capacity": "1.89 Ah",
"nominal_voltage": "3.7V",
"test_period": "2008",
"cycles_tested": "168",
"validation_role": "Independent test set for model validation"
}
}
]
created_assets = []
for config in nasa_configs:
print(f"\n📦 Processing {config['battery']}...")
try:
# Load NASA data
data_path = f"./data/nasa_converted/{config['battery']}.csv"
df = pd.read_csv(data_path)
if len(df) == 0:
print(f" ⚠️ No data found for {config['battery']}")
continue
# Get initial capacity
initial_capacity = df['capacity'].max()
# Generate predictions at different cycle points for historical data
cycle_points = [
int(len(df) * 0.25), # 25% through lifecycle
int(len(df) * 0.50), # 50% through lifecycle
int(len(df) * 0.75), # 75% through lifecycle
len(df) - 1 # Latest
]
# Create asset
asset_id = str(uuid.uuid4())
asset = Asset(
id=asset_id,
owner_id=config['owner'],
project_name=config['project_name'],
asset_id=config['asset_id'],
location=config['location'],
capacity=round(initial_capacity, 2),
bms_provider=config['bms_provider'],
category=json.dumps(config['category']),
project_stage=config['stage'],
is_online=1,
created_at=datetime.utcnow() - timedelta(days=90), # 3 months ago
updated_at=datetime.utcnow()
)
# Generate predictions for historical timeline
predictions = []
latest_prediction = None
for i, cycle_idx in enumerate(cycle_points):
try:
# Get data up to this cycle
data_subset = df.iloc[:cycle_idx+1]
# Run prediction
result = predict_soh(
battery_data=data_subset,
initial_capacity=initial_capacity,
forecast_horizons=[10, 50, 100]
)
# Calculate prediction date (spread over last 3 months)
days_ago = 90 - (i * 30) # 90, 60, 30, 0 days ago
pred_date = datetime.utcnow() - timedelta(days=days_ago)
# Create historical prediction
pred = HealthPrediction(
asset_id=asset_id,
prediction_date=pred_date,
cycle_index=result['current_cycle'],
health_score=result['health_score'],
current_soh=result['current_soh']['point_estimate'],
soh_lower_95=result['current_soh']['lower_95'],
soh_upper_95=result['current_soh']['upper_95'],
remaining_life_years=result['remaining_life']['years'],
remaining_life_cycles=result['remaining_life']['cycles'],
degradation_rate=result['degradation_rate']['per_year_pct'],
confidence=result['confidence'],
data_quality=result['data_quality']['quality_score'],
shap_top_features=json.dumps(
result.get('shap_explanation', {}).get('top_features', [])[:5]
) if result.get('shap_explanation') else None
)
predictions.append(pred)
latest_prediction = result
print(f" ✓ Prediction at cycle {cycle_idx}: Health={result['health_score']:.1f}/100")
except Exception as e:
print(f" ⚠️ Error at cycle {cycle_idx}: {e}")
continue
# Update asset with latest metrics
if latest_prediction:
asset.health_score = latest_prediction['health_score']
asset.remaining_life = latest_prediction['remaining_life']['years']
asset.degradation_rate = latest_prediction['degradation_rate']['per_year_pct']
asset.confidence = latest_prediction['confidence']
asset.last_prediction_date = datetime.utcnow()
# Add some operational data
asset.state_of_charge = 75.0 + (len(predictions) * 5) % 30
asset.voltage = 700.0 + (len(predictions) * 10) % 100
asset.temperature = 20.0 + (len(predictions) * 2) % 10
# Save to database
db.add(asset)
for pred in predictions:
db.add(pred)
db.commit()
db.refresh(asset)
# Store asset info before session closes
asset_info = {
"project_name": asset.project_name,
"asset_id": asset.asset_id,
"health_score": asset.health_score,
"remaining_life": asset.remaining_life,
"predictions_count": len(predictions)
}
created_assets.append(asset_info)
print(f" ✅ Created asset: {asset_info['project_name']}")
print(f" Health: {asset_info['health_score']:.1f}/100")
print(f" RUL: {asset_info['remaining_life']:.1f} years")
print(f" Historical predictions: {asset_info['predictions_count']}")
except Exception as e:
print(f" ❌ Error processing {config['battery']}: {e}")
db.rollback()
continue
print(f"\n✅ Seed complete! Created {len(created_assets)} assets")
print("\nAssets:")
for asset_info in created_assets:
print(f" - {asset_info['project_name']} ({asset_info['asset_id']}): Health={asset_info['health_score']:.1f}/100")
db.close()
return created_assets
if __name__ == "__main__":
seed_demo_assets()
|