dss_checks.py

"""
OpenDSS system checking functions for voltage violations, overloads, and blown fuses
"""

import opendssdirect as dss
import math
from config import VOLTAGE_UPPER_BOUND, VOLTAGE_LOWER_BOUND, TRANSFORMER_LOADING_THRESHOLD, LINE_LOADING_THRESHOLD


def check_blown_fuses():
    """Check for blown fuses in the system"""
    blown = []
    # If the circuit has no fuses, bail out cleanly
    if dss.Fuses.Count() == 0:
        return blown

    i = dss.Fuses.First()
    while i > 0:
        name = dss.Fuses.Name()
        try:
            if dss.Fuses.IsBlown():
                blown.append(name)
                print(name, "blown")
        except Exception as e:
            # Optional: log and continue if a specific fuse can't be activated
            print(f"Fuse check error on {name}: {e}")
        i = dss.Fuses.Next()
    return blown


def check_voltage_violations(ub=VOLTAGE_UPPER_BOUND, lb=VOLTAGE_LOWER_BOUND):
    """Check for voltage violations (over/under voltage)"""
    overvoltages_dict = {}
    undervoltages_dict = {}
    
    bus_names = dss.Circuit.AllBusNames()
    for b in bus_names:
        dss.Circuit.SetActiveBus(b)
        vang = dss.Bus.puVmagAngle()
        maxv = max(vang[::2])
        minv = min(vang[::2])
        if maxv > ub:
            overvoltages_dict[b] = maxv
        if minv < lb:
            undervoltages_dict[b] = minv
    
    return undervoltages_dict, overvoltages_dict


def check_xfmr_overloads(ub=TRANSFORMER_LOADING_THRESHOLD):
    """Check for transformer overloads"""
    transformer_violation_dict = {}
    unloaded_transformers_dict = {}
    
    dss.Circuit.SetActiveClass("Transformer")
    flag = dss.ActiveClass.First()
    while flag > 0:
        # Get the name of the Transformer
        transformer_name = dss.CktElement.Name()
        
        hs_kv = float(dss.Properties.Value('kVs').split('[')[1].split(',')[0])
        kva = float(dss.Properties.Value('kVA'))
        n_phases = dss.CktElement.NumPhases()
        
        if n_phases > 1:
            transformer_limit_per_phase = kva / (hs_kv * math.sqrt(3))
        else:
            transformer_limit_per_phase = kva / hs_kv
    
        primary_bus = dss.Properties.Value("buses").split('[')[1].split(',')[0]
        
        Currents = dss.CktElement.CurrentsMagAng()[:2*n_phases]
        Current_magnitude = Currents[::2]    
        
        transformer_current = Current_magnitude
        
        # Compute the loading
        ldg = max(transformer_current) / transformer_limit_per_phase
        
        # If the loading is more than threshold, store the violation
        if ldg > ub:
            transformer_violation_dict[transformer_name] = {
                'Bus': primary_bus, 
                'Loading (p.u.)': ldg, 
                'kVA': kva, 
                'number_of_phases': n_phases
            }
        elif ldg == 0:
            unloaded_transformers_dict[transformer_name] = {
                'Bus': primary_bus, 
                'kVA': kva, 
                'number_of_phases': n_phases
            }
        
        # Move on to the next Transformer...
        flag = dss.ActiveClass.Next()
    
    return transformer_violation_dict, unloaded_transformers_dict


def check_line_overloads(ub=LINE_LOADING_THRESHOLD):
    """Check for line overloads"""
    line_overloads_dict = {}
    unloaded_line_dict = {}
    
    # Set the active class to be the lines
    dss.Circuit.SetActiveClass("Line")

    # Loop over the lines
    flag = dss.ActiveClass.First()
    while flag > 0:
        line_name = dss.CktElement.Name()
        # Get the current limit
        bus1 = dss.Properties.Value("bus1")
        bus2 = dss.Properties.Value("bus2")
        line_limit_per_phase = dss.CktElement.NormalAmps()
        
        # Compute the current through the line
        phase = int(.25 * len(dss.CktElement.Currents()))
        line_current = dss.CktElement.CurrentsMagAng()[:2*phase]
        line_current = line_current[::2]
        
        # The loading is the ratio of the two
        ldg = max(line_current) / float(line_limit_per_phase)
        
        if ldg > ub:
            line_overloads_dict[line_name] = {
                'Bus1': bus1, 
                'Bus2': bus2, 
                'Loading (p.u.)': ldg
            }
        elif ldg == 0:
            unloaded_line_dict[line_name] = {'Bus1': bus1, 'Bus2': bus2}
        
        # Move on to the next line
        flag = dss.ActiveClass.Next()
        
    return line_overloads_dict, unloaded_line_dict