Resources

from ona_sdk import OnaClient

# Initialize client
client = OnaClient("your-api-key")

# Upload historical data
upload_result = client.ingest_historical(
    customer_id="solar-farm-001",
    file_path="historical_data.csv",
    manufacturer="SolarEdge",
    location="CapeTown"
)

# Generate 24-hour forecast
forecast = client.generate_forecast(
    customer_id="solar-farm-001",
    forecast_horizon=24,
    model_type="lstm"
)

# Get results
results = client.get_forecast_results(
    customer_id="solar-farm-001",
    forecast_id=forecast["forecast_id"]
)

print(f"Forecast accuracy: {results['accuracy']}%")

import asyncio
from ona_sdk import OnaClient

async def process_multiple_sites():
    client = OnaClient("your-api-key")
    
    sites = [
        {"id": "site-001", "location": "CapeTown"},
        {"id": "site-002", "location": "Johannesburg"},
        {"id": "site-003", "location": "Durban"}
    ]
    
    # Process all sites concurrently
    tasks = []
    for site in sites:
        task = client.generate_forecast(
            customer_id=site["id"],
            forecast_horizon=48
        )
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    
    # Aggregate results
    total_forecast = sum(r["total_power"] for r in results)
    print(f"Total forecasted power: {total_forecast} MW")

import asyncio
from ona_sdk import OnaClient

async def stream_solar_data():
    client = OnaClient("your-api-key")
    
    # Set up webhook for notifications
    webhook = await client.create_webhook(
        url="https://your-app.com/webhooks/solar-data",
        events=["data.ingested", "forecast.completed"]
    )
    
    # Stream real-time data
    async for data_point in client.stream_realtime_data():
        # Process incoming data
        processed_data = {
            "timestamp": data_point["timestamp"],
            "power": data_point["power"],
            "efficiency": data_point["power"] / data_point["capacity"]
        }
        
        # Send to forecasting API
        await client.ingest_nowcast(
            customer_id="solar-farm-001",
            data=processed_data
        )

from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)

@app.route('/webhooks/ona', methods=['POST'])
def ona_webhook():
    # Verify webhook signature
    signature = request.headers.get('X-Ona-Signature')
    payload = request.get_data()
    
    if not verify_signature(payload, signature, 'your-webhook-secret'):
        return jsonify({'error': 'Invalid signature'}), 401
    
    # Process webhook event
    event = request.json
    
    if event['event'] == 'forecast.completed':
        # Handle forecast completion
        forecast_id = event['data']['forecast_id']
        customer_id = event['data']['customer_id']
        
        # Update your application state
        update_dashboard(customer_id, forecast_id)
        
    elif event['event'] == 'data.ingested':
        # Handle new data ingestion
        process_new_data(event['data'])
    
    return jsonify({'status': 'success'})

def verify_signature(payload, signature, secret):
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(f"sha256={expected}", signature)

if __name__ == '__main__':
    app.run(debug=True)