Automating Power Systems Engineering Workflows: From Scripts to Production

Power systems engineering is drowning in repetitive work. Running the same contingency analysis for the 47th time. Manually copying results into Word templates. Re-running studies because base-case data changed. Waiting days for batch jobs when you need answers in hours.

This doesn't have to be your reality. Modern power systems automation can turn weeks of manual work into minutes of computation—if you know how to build it correctly.

1. What Power Systems Automation Really Is

Power systems automation is not "run a Python script once." It's closer to:

• A data pipeline (fetch base case → validate → run studies → aggregate results → generate reports)
• A reliability layer (retries, error handling, logging, alerting)
• An integration fabric (PSS®E + PSCAD + Excel + databases + ISO APIs + email)
• A compliance framework (version control, audit trails, reproducibility)

The key mental model: you're not just automating a task—you're building a system that runs unattended and produces trustworthy results.

2. The Automation Toolbox

Modern power systems automation leverages several categories of tools:

2.1 Simulation Software APIs

• PSS®E (psspy): Python API for Siemens PSS®E, embedded Python interpreter
• PSCAD Automation Library: Python-based interface for batch PSCAD studies
• PowerWorld Automation Server: COM interface for automated dispatch studies
• TARA: Python scripting for transient analysis workflows

2.2 Open-Source Python Libraries

• PyPSA: Framework for optimizing modern power systems (1.8K stars, actively maintained)
• pandapower: Power flow, OPF, state estimation, short circuit (IEC 60909)
• Grid2Op: Reinforcement learning for grid operations
• PyPSSE (NREL): Wrapper around psspy for easier scripting

2.3 Orchestration & Scheduling

• Apache Airflow: Python-based DAG orchestration for complex workflows
• Windows Task Scheduler / cron: OS-level job scheduling
• GitHub Actions / GitLab CI: Git-triggered automation for model validation

2.4 Data Integration

• requests / httpx: REST API clients for ISO data feeds
• pandas: Time-series data manipulation (SCADA, market prices)
• SQLAlchemy: Database ORM for persistent result storage
• FastAPI / Flask: Build webhooks for event-driven automation

3. PSS®E Python Automation (psspy)

PSS®E ships with an embedded Python interpreter and the psspy module. This is your primary interface for automating load flow, dynamics, and contingency analysis.

3.1 Basic PSS®E Automation Pattern

import os
import sys

# Add PSSE to path (adjust version as needed)
PSSE_PATH = r"C:\Program Files\PTI\PSSE35\PSSPY27"
sys.path.append(PSSE_PATH)
os.environ['PATH'] += f";{PSSE_PATH}"

import psspy
import redirect

# Suppress output (optional)
redirect.psse2py()

# Initialize PSS/E
psspy.psseinit(150000)  # 150k bus limit

# Load base case
ierr = psspy.case(r"C:\Cases\BaseCase_2026.sav")
if ierr != 0:
    raise RuntimeError(f"Failed to load case, error code: {ierr}")

# Run power flow
psspy.fnsl([0, 0, 0, 1, 1, 0, 99, 0])

# Extract results
ierr, bus_voltages = psspy.abusreal(string='PU')

# Process and report
for bus_id, voltage in zip(range(len(bus_voltages[0])), bus_voltages[0]):
    if voltage < 0.95 or voltage > 1.05:
        print(f"Bus {bus_id}: {voltage:.3f} p.u. - OUT OF RANGE")

print("Analysis complete.")

3.2 Production-Grade PSS®E Script Structure

For reliable automation, wrap psspy calls in proper error handling and logging:

import logging
from pathlib import Path
from typing import Dict, Any

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('psse_automation.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class PSSEAutomation:
    def __init__(self, case_path: Path):
        self.case_path = case_path
        self._initialize_psse()

    def _initialize_psse(self):
        """Initialize PSS/E with error handling"""
        try:
            import psspy
            psspy.psseinit(150000)
            logger.info("PSS/E initialized successfully")
        except Exception as e:
            logger.error(f"Failed to initialize PSS/E: {e}")
            raise

    def load_case(self) -> None:
        """Load case with validation"""
        if not self.case_path.exists():
            raise FileNotFoundError(f"Case file not found: {self.case_path}")

        ierr = psspy.case(str(self.case_path))
        if ierr != 0:
            raise RuntimeError(f"Failed to load case: {self.case_path}, error: {ierr}")

        logger.info(f"Loaded case: {self.case_path.name}")

    def run_contingency_analysis(self, contingencies: list) -> Dict[str, Any]:
        """Run N-1 contingencies with results aggregation"""
        results = {'violations': [], 'summary': {}}

        for contingency in contingencies:
            logger.info(f"Running contingency: {contingency['name']}")
            # Implement contingency logic here
            # ...

        return results

    def generate_report(self, results: Dict[str, Any], output_path: Path) -> None:
        """Generate HTML/PDF report from results"""
        # Report generation logic
        pass

# Usage
if __name__ == "__main__":
    automation = PSSEAutomation(Path("BaseCase_2026.sav"))
    automation.load_case()
    results = automation.run_contingency_analysis([...])
    automation.generate_report(results, Path("report.html"))

4. PSCAD Automation & Batch Processing

PSCAD's Automation Library (AL) enables Python-based workflow automation for EMT studies—critical for weak grid interconnections and IBR validation.

4.1 PSCAD Automation Library Basics

from pscad import *

# Initialize PSCAD instance
pscad_instance = pscad.PSCAD()
pscad_instance.load_workspace("C:/Projects/EMT_Study.pswx")

# Load project
project = pscad_instance.project("GridFormingBESS")

# Modify parameters
project.user_cmp("IBR_Control").parameters("Droop_Gain").value = 0.05

# Run simulation
project.run()

# Wait for completion
while project.is_running():
    time.sleep(1)

# Extract results
results = project.output_channel("POI_Voltage").get_data()

# Process and save
import pandas as pd
df = pd.DataFrame(results, columns=['Time', 'Voltage'])
df.to_csv("voltage_results.csv", index=False)

print(f"Simulation complete. Peak voltage: {df['Voltage'].max():.3f} p.u.")

4.2 Batch Parameter Sweeps for Sensitivity Analysis

import itertools
from concurrent.futures import ProcessPoolExecutor

# Define parameter ranges
scr_values = [1.5, 2.0, 2.5, 3.0]
droop_gains = [0.03, 0.04, 0.05, 0.06]

# Generate all combinations
param_combinations = list(itertools.product(scr_values, droop_gains))

def run_single_case(params):
    """Run a single PSCAD case with given parameters"""
    scr, droop = params

    # Initialize PSCAD
    pscad_inst = pscad.PSCAD()
    pscad_inst.load_workspace("EMT_Study.pswx")
    project = pscad_inst.project("Sensitivity")

    # Set parameters
    project.user_cmp("Grid").parameters("SCR").value = scr
    project.user_cmp("IBR").parameters("Droop").value = droop

    # Run
    project.run()
    while project.is_running():
        time.sleep(0.5)

    # Extract stability metric
    voltage = project.output_channel("POI_V").get_data()
    max_overshoot = max(voltage) - 1.0

    return {'SCR': scr, 'Droop': droop, 'Overshoot': max_overshoot}

# Run in parallel (careful with PSCAD licenses!)
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(run_single_case, param_combinations))

# Aggregate and analyze
import pandas as pd
df = pd.DataFrame(results)
print(df.pivot_table(values='Overshoot', index='SCR', columns='Droop'))

5. Production Workflow Patterns

Pattern A: "Nightly Base Case Update"

Use when: You need fresh base cases from ISO portals daily

# nightly_basecase_update.py
import requests
from pathlib import Path
from datetime import datetime
import hashlib

def download_iso_basecase(url: str, output_path: Path) -> bool:
    """Download base case with retry logic"""
    for attempt in range(3):
        try:
            response = requests.get(url, timeout=300)
            response.raise_for_status()

            output_path.write_bytes(response.content)

            # Validate checksum if provided
            checksum = response.headers.get('X-Checksum')
            if checksum:
                actual = hashlib.md5(response.content).hexdigest()
                if actual != checksum:
                    raise ValueError(f"Checksum mismatch: {actual} != {checksum}")

            logger.info(f"Downloaded base case: {output_path.name}")
            return True
        except Exception as e:
            logger.warning(f"Download attempt {attempt + 1} failed: {e}")

    return False

def validate_and_convert(raw_path: Path, psse_path: Path) -> bool:
    """Validate and convert to PSS/E format"""
    # Conversion logic here
    # ...
    return True

def test_convergence(case_path: Path) -> bool:
    """Test if case converges in PSS/E"""
    try:
        psspy.case(str(case_path))
        ierr = psspy.fnsl([0, 0, 0, 1, 1, 0, 99, 0])
        return ierr == 0
    except Exception as e:
        logger.error(f"Convergence test failed: {e}")
        return False

# Main workflow
iso_url = "https://oasis.spp.org/cases/latest.raw"
timestamp = datetime.now().strftime("%Y%m%d")
raw_path = Path(f"raw_cases/SPP_{timestamp}.raw")
psse_path = Path(f"validated_cases/SPP_{timestamp}.sav")

if download_iso_basecase(iso_url, raw_path):
    if validate_and_convert(raw_path, psse_path):
        if test_convergence(psse_path):
            # Commit to version control
            os.system(f'git add {psse_path}')
            os.system(f'git commit -m "Auto: Base case {timestamp}"')

            # Trigger dependent studies
            trigger_webhook("https://automation.gridopt.io/trigger/interconnection-study")
        else:
            send_alert("Convergence test failed", severity="high")

Pattern B: "Event-Driven Interconnection Study"

Use when: New interconnection request arrives in queue

Pattern C: "Scheduled Compliance Reports"

Use when: Weekly/monthly NERC compliance reports needed

6. Scheduling & Orchestration

6.1 Windows Task Scheduler (Simple Cron)

For basic scheduled tasks on Windows:

# Create a scheduled task via PowerShell
$action = New-ScheduledTaskAction -Execute "python.exe" -Argument "C:\Automation\nightly_update.py"
$trigger = New-ScheduledTaskTrigger -Daily -At 2:00AM
$principal = New-ScheduledTaskPrincipal -UserId "DOMAIN\ServiceAccount" -LogonType Password

Register-ScheduledTask -TaskName "BaseCase_Update" -Action $action -Trigger $trigger -Principal $principal

6.2 Apache Airflow (Complex DAGs)

For complex dependencies and monitoring:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'gridopt',
    'depends_on_past': False,
    'start_date': datetime(2026, 1, 1),
    'email': ['[email protected]'],
    'email_on_failure': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'interconnection_study_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # Daily at 2 AM
    catchup=False
)

download_task = PythonOperator(
    task_id='download_basecase',
    python_callable=download_iso_basecase,
    dag=dag
)

validate_task = PythonOperator(
    task_id='validate_case',
    python_callable=validate_and_convert,
    dag=dag
)

powerflow_task = PythonOperator(
    task_id='run_powerflow',
    python_callable=run_psse_powerflow,
    dag=dag
)

contingency_task = PythonOperator(
    task_id='run_contingencies',
    python_callable=run_n1_analysis,
    dag=dag
)

report_task = PythonOperator(
    task_id='generate_report',
    python_callable=generate_html_report,
    dag=dag
)

# Define dependencies
download_task >> validate_task >> powerflow_task >> contingency_task >> report_task

7. ISO Data Integration & Webhooks

7.1 Fetching OASIS Data

import requests
from datetime import datetime
import pandas as pd

def fetch_spp_lmp_data(start_date: str, end_date: str) -> pd.DataFrame:
    """Fetch SPP real-time LMP data from OASIS API"""

    url = "https://marketplace.spp.org/api/rtbm-lmp"
    params = {
        'startDate': start_date,
        'endDate': end_date,
        'format': 'json'
    }

    response = requests.get(url, params=params, timeout=60)
    response.raise_for_status()

    data = response.json()
    df = pd.DataFrame(data['records'])

    # Process timestamps
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['lmp'] = pd.to_numeric(df['lmp'])

    return df

# Usage in automation
lmp_data = fetch_spp_lmp_data('2026-01-01', '2026-01-31')
print(f"Fetched {'{'} len(lmp_data){'}'} LMP records")
print(f"Avg LMP: { lmp_data['lmp'].mean():.2f {'}'} /MWh")

7.2 Building Webhooks for Event-Driven Automation

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import subprocess

app = FastAPI()

class InterconnectionRequest(BaseModel):
    queue_number: str
    capacity_mw: float
    location: str
    interconnection_voltage_kv: float

@app.post("/webhook/new-interconnection")
async def handle_new_interconnection(request: InterconnectionRequest):
    """Webhook endpoint for new interconnection requests"""

    # Validate request
    if request.capacity_mw <= 0:
        raise HTTPException(status_code=400, detail="Invalid capacity")

    # Trigger automated study
    result = subprocess.run([
        'python',
        'run_interconnection_study.py',
        '--queue', request.queue_number,
        '--capacity', str(request.capacity_mw),
        '--location', request.location
    ], capture_output=True, text=True)

    if result.returncode != 0:
        raise HTTPException(status_code=500, detail=f"Study failed: {result.stderr}")

    return {
        "status": "success",
        "queue_number": request.queue_number,
        "message": "Interconnection study initiated"
    }

# Run with: uvicorn webhook_server:app --host 0.0.0.0 --port 8000

8. Automated Compliance & Validation

8.1 MOD-026-2 Model Cross-Validation

def validate_emt_rms_crosscheck(emt_results: pd.DataFrame, rms_results: pd.DataFrame) -> dict:
    """
    Automated validation for MOD-026-2 compliance
    Compare EMT vs RMS model responses for same disturbance
    """

    # Align time series
    emt_voltage = emt_results['POI_Voltage']
    rms_voltage = rms_results['POI_Voltage']

    # Calculate correlation
    correlation = emt_voltage.corr(rms_voltage)

    # Calculate RMSE
    rmse = np.sqrt(np.mean((emt_voltage - rms_voltage)**2))

    # Check peak deviation
    peak_diff = abs(emt_voltage.max() - rms_voltage.max())

    compliance_status = {
        'correlation': correlation,
        'rmse': rmse,
        'peak_deviation': peak_diff,
        'passes': correlation > 0.95 and rmse < 0.05 and peak_diff < 0.1
    }

    if not compliance_status['passes']:
        logger.warning(f"MOD-026-2 validation failed: {compliance_status}")

    return compliance_status

8.2 Automated Base Case Quality Checks

def run_basecase_qa(case_path: Path) -> dict:
    """Run automated quality assurance on base case"""

    issues = []

    # Load case
    psspy.case(str(case_path))

    # Check 1: Voltage violations
    ierr, voltages = psspy.abusreal(string='PU')
    for i, v in enumerate(voltages[0]):
        if v < 0.90 or v > 1.10:
            issues.append(f"Bus {i}: Voltage {v:.3f} p.u. out of range")

    # Check 2: Islanded buses
    ierr, islands = psspy.tree(1, 1)
    if islands > 1:
        issues.append(f"Found {islands} islands in network")

    # Check 3: Missing generator data
    ierr, gen_count = psspy.amachcount()
    if gen_count == 0:
        issues.append("No generators found in case")

    # Check 4: Overloaded branches
    ierr, flows = psspy.aflowreal(string='PCTRATE')
    for i, flow in enumerate(flows[0]):
        if abs(flow) > 100:
            issues.append(f"Branch {i}: {flow:.1f}% loading")

    return {
        'valid': len(issues) == 0,
        'issue_count': len(issues),
        'issues': issues[:10]  # First 10 issues
    }

9. CI/CD for Power Systems Models

Treat power systems models like software: version control + automated testing + deployment pipelines.

9.1 Git Workflow for Base Cases

# .gitattributes
*.sav binary
*.raw text
*.dyr text

# Track large files with Git LFS
git lfs track "*.sav"
git lfs track "*.out"

9.2 GitHub Actions for Model Validation

# .github/workflows/validate-basecase.yml
name: Validate Base Case

on:
  push:
    paths:
      - 'cases/*.sav'
      - 'cases/*.raw'

jobs:
  validate:
    runs-on: windows-latest

    steps:
      - uses: actions/checkout@v3

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.8'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt

      - name: Run convergence test
        run: |
          python scripts/test_convergence.py cases/BaseCase_2026.sav

      - name: Run QA checks
        run: |
          python scripts/run_qa_checks.py cases/BaseCase_2026.sav

      - name: Generate validation report
        run: |
          python scripts/generate_report.py --output validation_report.html

      - name: Upload report
        uses: actions/upload-artifact@v3
        with:
          name: validation-report
          path: validation_report.html

10. Debugging, Reliability, and Common Pitfalls

10.1 Common Pitfalls

10.2 Reliability Checklist

• Idempotency: Can the script run multiple times safely?
• Error handling: Does it fail gracefully with clear error messages?
• Logging: Can you debug issues from logs alone?
• Validation: Does it check inputs and outputs?
• Monitoring: Will you know if it fails at 3 AM?
• Documentation: Can someone else run it 6 months from now?

11. Reference Architectures

Architecture A: "Centralized Study Server"

Architecture B: "Cloud-Native Batch Processing"

Architecture C: "Hybrid On-Prem + Cloud"