""" 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()