TextModel.py

'''TextModel.py

Provides a re-implementation of the OpenDSS model, relying on 

    dss.DSS.Text.Command
    
and

    dss.DSS.Text.Result

more than our implementation in System.Model.

'''

import os
import random
import pickle
import dss
from time import time
from math import sin, cos, sqrt, atan, pi

import numpy as np
import pandas as pd


_cwd = os.getcwd()

# TODO: Dataclass to store voltages, currents, powers, taps
from dataclasses import dataclass

# TODO: Consider a dataclass to store timeline of fixed length, size

'''TODOs
1. Reimplement init
2. Reimplement read
3. Test model
4. Test if dataclasses are faster than dicts
5. Dataclass for data, timeline
'''

class TextModel:
    ''' Provides a wrapper to the OpenDSS circuit.

    Instantiates an OpenDSS circuit, adds custom loads.
    With each `step` command, the simulation iterates, updating
    self.grid_timeline, a list indexed by timestamp,
        of dict[string element_id]
            of dict[string property_name]
                of number of NumpyArray.
    

    :param cwd: Operating directory, defaults to _cwd
    :type cwd: str, optional
    :param dssfolder: Path to load OpenDSS model from,
        defaults to f"{_cwd}dss_model_with_waterheaters/"
    :type dssfolder: str, optional
    :param redirect_file:
    :type redirect_file:
    :param elements_to_read:
    :type elements_to_read:
    :param precision:
    :type precision:
    '''
    def __init__(self,
        cwd = _cwd,
        dssfolder = f"{_cwd}/other_DSS/13Bus/",
        redirect_file = 'run.dss',
        elements_to_read = ['transformer.reg1'],
        precision = np.float32,
    ):

        self.cwd = cwd
        self.dssfolder = dssfolder
        self.redirect_file = redirect_file
        self.elements_to_read = elements_to_read
        self.precision = precision

        self.engine = dss.DSS
        self.engine.Start(0)
        self.textcommand(
            f"compile {self.redirect_file}"
        )
        self.circuit = self.engine.ActiveCircuit
        
        self.grid_state = self.read() # TODO
        self.grid_timeline = [self.grid_state]
    # TODO
    
    def restart(self,):
        """Restart the engine, required for concurrent runs
        """
        self.engine.Start(0)
        self._run_command("reset")
        self._run_command("clear")
        # see also: engine.ClearAll(), engine.Reset()
        self.engine.Reset()
        self.textcommand(
            f"compile {self.redirect_file}")
        self.circuit = self.engine.ActiveCircuit

    def _repl(self):
        '''Start an interactive OpenDSS prompt. Quit with 'exit'.'''
        command = input("--> ").lower()
        while not command == 'exit':
            print(self.textcommand(command))
            command = input("--> ").lower()
    
    def textcommand(self, dss_command):
        """Run OpenDSS command using raw string manipulation.

        OpenDSS stores the results of text commands in self.engine.Text.Result
        e.g. self._run_command("help") prints out some help information
             and returns the empty string ""
        e.g. self._run_command("? transformer.reg3a.tap")  returns '1.0125'
        
        :param dss_command: The DSS string to be run.
        :type dss_command: String
        
        :return: The string result after running the text command
        :rtype: String
        """
        self.engine.Text.Command = dss_command
        return self.engine.Text.Result
    

    def set_load(self, load_name, kW = 0):
        """Convenience function to set a load.

        :param load_name: Element name corresponding to load, e.g. 'load.wh'
        :type load_name: string
        :param kW: Wattage to set the load to, defaults to 0
        :type kW: int, optional
        :return: Empty string when run properly
        :rtype: String
        """

        return self.textcommand(f"load.{load_name}.kW = {kW}")
        # model.set_load("waterheater1", 1234)
        # load.waterheater1.kW = 1234

    def read_element(self, element_name):
        '''Set the active element and read relevant properties from it.

        :param element_name:
        :type element_name:

        :returns: Dictionary mapping 
        :rtype: dict

        Set the active circuit and read voltages, currents, powers, and taps (if relevant)
        '''
        # e.g. 'Transformer.Reg1' becomes eclass = 'transformer', ename = 'reg1' 
        eclass, ename = element_name.strip().split(".")
        eclass = eclass.strip().lower()
        ename = ename.strip().lower()

        # set the active element to be read from
        self.textcommand(f"select {eclass}.{ename}")

        properties = {}

        # try voltages
        voltages = self.textcommand(f"voltages")
        if len(voltages) > 0:
            properties['V'] = _interpret_line(voltages, precision=self.precision)

        # try currents
        currents = self.textcommand(f"currents")
        if len(currents) > 0:
            properties['I'] = _interpret_line(currents, precision=self.precision)

        # try powers
        powers = self.textcommand(f"powers")
        if len(powers) > 0:
            properties['S'] = _interpret_line(powers, precision=self.precision)

        # try taps
        # todo -- do this using strings?
        if eclass == "transformer":
            active_element = self.circuit.ActiveElement
            properties['tap'] = float(active_element.Properties('tap').Val)

        return properties

    def read(self, elements_to_read = None):
        """Given a list of elements to read (defaulting to
        self.elements_to_read if not specified), return the state of
        all of those elements.

        :param elements_to_read: List of string, defaults to None
        :type elements_to_read: list[str], optional
        :return: [description]
        :rtype: [type]
        """
        grid_state = {}
        if elements_to_read is None:
            elements_to_read = self.elements_to_read

        for element_name in elements_to_read:
            element_properties = self.read_element(element_name)
            grid_state[element_name.lower()] = element_properties
        
        return grid_state

    def step(self):
        """Move through one iteration of simulation, updating the grid state timeline.

        :return: String result of running 'solve'
        :rtype: str
        """
        result = self.textcommand('solve')
        self.grid_state = self.read()
        self.grid_timeline.append(self.grid_state)
        return result
    
    def dump_to_pickle(self, path):
        """Save the grid timeline as a pickle.

        :param path: Path to save pickle to
        :type path: str
        """
        if path is None:
            path = self.cwd + "textmodel_output.pkl"
        
        pd.to_pickle(self.grid_timeline, path)




def _interpret_line(line, precision = np.float32):
    '''Interpret the result of a DSS text command. 
    :param line: String, e.g. '   1123.1    120.0 '  
    :type line: String
    :param precision:
    :type precision:
    
    :returns: List of number
    :type: np.ndarray
    '''
    # handle case where line is totally empty
    values = []

    if len(line) > 0: # returns empty array if not
        for element in line.strip().split(','):
            raw_element = element.strip(" ,")
            # handle case where string is empty
            if len(raw_element) > 0:
                values.append(float(raw_element))
    
    return np.array(values, dtype = precision)