Complete Workflow

The complete_workflow_example.py is the capstone example for the Ona Intelligence Layer SDK. It demonstrates how to chain multiple services together into a single end-to-end workflow: from solar forecasting through fault detection, diagnostics, and maintenance scheduling.

Source: complete_workflow_example.py

Workflow Overview

The example follows a 7-step pipeline that mirrors a real operational scenario:

1. Get Solar Forecast — Retrieve the next 24-hour production forecast
2. Run Fault Detection — Analyze an inverter for anomalies
3. Run Diagnostics — If a fault is detected, diagnose the root cause
4. Create Maintenance Schedule — Schedule a work order based on the diagnosis
5. Review Activities — List recent terminal activities
6. Check ML Model Registry — Inspect registered detection models
7. Get Nowcast Data — Pull real-time site metrics

Initialization

The workflow begins by initializing a single OnaClient instance. This client provides access to all services — no separate initialization is needed per service.

from ona_platform import OnaClient

def main():
    """Demonstrate a complete workflow across multiple services."""

    # Initialize client
    client = OnaClient()
    customer_id = "customer123"
    site_id = "Sibaya"
    asset_id = "INV001"

    print("=== COMPLETE WORKFLOW EXAMPLE ===\n")

The OnaClient constructor reads endpoint URLs and API keys from environment variables. The same client object exposes .forecasting, .terminal, and other service attributes.

Step 1: Get Solar Forecast

Retrieve the site-level production forecast for the next 24 hours. The forecasting service returns predicted kWh output per time interval.

    # Step 1: Get current forecast
    print("Step 1: Get Solar Forecast")
    forecast = client.forecasting.get_site_forecast(
        site_id=site_id,
        forecast_hours=24
    )
    print(f"  - Site: {forecast['site_id']}")
    print(f"  - Devices: {forecast['devices_included']}")
    print(f"  - Next hour forecast: {forecast['forecasts'][0]['kWh_forecast']} kWh")

Expected output:

Step 1: Get Solar Forecast
  - Site: Sibaya
  - Devices: 12
  - Next hour forecast: 45.2 kWh

The forecast response includes site_id, devices_included, and a forecasts array where each entry contains a timestamp, kWh_forecast value, hour_ahead count, and a confidence score.

Step 2: Run Fault Detection

Pass the inverter asset ID to the terminal service’s detection endpoint. This triggers an analysis over the last 6 hours of telemetry data.

    # Step 2: Check asset health via detection
    print("\nStep 2: Run Fault Detection")
    detection = client.terminal.run_detection(
        customer_id=customer_id,
        asset_id=asset_id,
        lookback_hours=6
    )
    detection_id = detection['detection_id']
    severity = detection['analysis']['severity_label']
    print(f"  - Detection ID: {detection_id}")
    print(f"  - Severity: {severity}")
    print(f"  - Fault type: {detection['analysis']['fault_type']}")
    print(f"  - Summary: {detection['analysis']['summary']}")

Expected output:

Step 2: Run Fault Detection
  - Detection ID: det_8f3a2b1c
  - Severity: High
  - Fault type: capacity_drop
  - Summary: Significant power drop detected on INV001 over the last 6 hours

The detection response includes a detection_id (needed for diagnostics) and an analysis object with severity_label (Low, Medium, or High), fault_type, and summary.

Step 3: Run Diagnostics (Conditional)

If the detection severity is High or Medium, the workflow escalates to diagnostics. This step takes the detection_id from Step 2 and performs a deeper root-cause analysis.

    # Step 3: If fault detected, run diagnostics
    if severity in ['High', 'Medium']:
        print("\nStep 3: Run Diagnostics (fault detected)")
        diagnostic = client.terminal.run_diagnostics(
            customer_id=customer_id,
            asset_id=asset_id,
            detection_id=detection_id,
            lookback_hours=6
        )
        print(f"  - Root cause: {diagnostic['analysis']['root_cause']}")
        print(f"  - Category: {diagnostic['analysis']['category']}")
        print(f"  - Confidence: {diagnostic['analysis']['confidence']}")
        for action in diagnostic['analysis'].get('recommended_actions', []):
            print(f"  - Action: {action}")

Expected output:

Step 3: Run Diagnostics (fault detected)
  - Root cause: DC string underperformance — possible soiling or partial shading
  - Category: string_fault
  - Confidence: 0.87
  - Action: Inspect DC connectors and wiring within 48 hours
  - Action: Check for vegetation growth or soiling on panels

The diagnostic response provides a human-readable root_cause, a machine-readable category, a confidence score (0–1), and a recommended_actions list with specific remediation steps.

Step 4: Create Maintenance Schedule

Based on the diagnostic root cause and severity, the workflow creates a maintenance schedule with an estimated duration.

        # Step 4: Create maintenance schedule
        print("\nStep 4: Create Maintenance Schedule")
        schedule = client.terminal.create_schedule(
            customer_id=customer_id,
            asset_id=asset_id,
            description=f"Maintenance: {diagnostic['analysis']['root_cause']}",
            priority="High" if severity == "High" else "Medium",
            estimated_duration_hours=8
        )
        print(f"  - Schedule created: {schedule['schedule_id']}")

Expected output:

Step 4: Create Maintenance Schedule
  - Schedule created: sched_a1b2c3d4

The create_schedule method accepts a description, priority (High or Medium), and estimated_duration_hours. It returns a schedule_id that can be tracked in the activity stream.

Step 5: Review Recent Activities

List the terminal activities from the last 24 hours. This includes detections, diagnostics, and schedules created in the current workflow.

    # Step 5: Review activity stream
    print("\nStep 5: Review Recent Activities")
    activities = client.terminal.list_activities(customer_id=customer_id)
    print(f"  - Total activities (24h): {len(activities)}")
    print("  - Recent activities:")
    for activity in activities[:3]:
        print(f"    [{activity['phase']}] {activity['title']}")

Expected output:

Step 5: Review Recent Activities
  - Total activities (24h): 7
  - Recent activities:
    [act] Maintenance schedule created
    [decide] Diagnostics completed — string_fault
    [observe] Anomaly detected on INV001

Each activity includes a phase (corresponding to the OODA loop: observe, orient, decide, act) and a title.

Step 6: Check ML Model Registry

Inspect the registered ML models used for detection and diagnostics. This is useful for auditing which model versions are active.

    # Step 6: Check ML model registry
    print("\nStep 6: Check ML Model Registry")
    models = client.terminal.get_ml_models()
    print(f"  - Registered models: {len(models)}")
    if models:
        latest_model = models[0]
        print(f"  - Latest: {latest_model.get('model_name', 'N/A')} "
              f"(version {latest_model.get('version', 'N/A')})")

Expected output:

Step 6: Check ML Model Registry
  - Registered models: 4
  - Latest: ona-anomaly-detector (version 2.1.3)

Step 7: Get Nowcast Data

Finally, pull real-time nowcast metrics for monitoring. This provides a snapshot of current site performance.

    # Step 7: Get nowcast data for monitoring
    print("\nStep 7: Get Nowcast Data")
    nowcast = client.terminal.get_nowcast_data(
        customer_id=customer_id,
        time_range="1h"
    )
    metrics = nowcast.get('data', {}).get('latest_metrics', {})
    print(f"  - Total power: {metrics.get('total_power_kw', 0)} kW")
    print(f"  - Active inverters: {metrics.get('active_inverters', 0)}/{metrics.get('total_inverters', 0)}")
    print(f"  - Producing: {metrics.get('producing_inverters', 0)}")
    print(f"  - Avg temperature: {metrics.get('avg_temperature_c', 0)}°C")

    print("\n=== WORKFLOW COMPLETE ===")


if __name__ == '__main__':
    main()

Expected output:

Step 7: Get Nowcast Data
  - Total power: 312.5 kW
  - Active inverters: 11/12
  - Producing: 11
  - Avg temperature: 42.3°C

=== WORKFLOW COMPLETE ===

The nowcast response wraps metrics inside a data envelope — access latest_metrics via nowcast['data']['latest_metrics']. The active_inverters count includes all online inverters; producing_inverters counts only those actively generating power.

Key Takeaways

• Single client, multiple services — One OnaClient instance provides access to forecasting, terminal, and all other services.
• Detection → Diagnostics chain — The detection_id from run_detection() is the input to run_diagnostics(), creating a natural pipeline.
• Conditional escalation — Diagnostics and scheduling only run when severity warrants it (High or Medium).
• Activity audit trail — Every action (detection, diagnostic, schedule) appears in the activity stream, tagged with its OODA phase.

For a comparison of how this same workflow looks in JavaScript, see Python vs JavaScript. For the full list of available examples, see Code Examples.