OpenDSS to LinDistRestoration Parser

DSSManager

ldrestoration.dssparser.dssparser.DSSManager

DSSManager is the primary module to parse the OpenDSS data. It manages all the components, including but not limited to, loads, generators, pdelements, transformers etc. Each of the components' data structure is managed by their respective handlers and can be accessed individually, if required.

Parameters:

Name	Type	Description	Default
dssfile	str	path of the dss master file (currently only supports OpenDSS files)	required
include_DERs	bool	Check whether to include DERs or not. Defaults to True.	False
DER_pf	float	Constant power factor of DERs. Defaults to 0.9.	0.9
include_secondary_network	bool	Check whether to include secondary network or not. Defaults to False.	False

Examples:

The only required argument is the OpenDSS master file. We assume that the master file compiles all other OpenDSS files. The DSSManager class is initiated first and a method parse_dss() will then parse the overall data.

>>> dataobj = DSSManager('ieee123master.dss', include_DERs=True)
>>> dataobj.parse_dss()

Source code in ldrestoration/dssparser/dssparser.py

class DSSManager:
    """DSSManager is the primary module to parse the OpenDSS data. It manages all the components, including but not limited to, loads, generators, pdelements, transformers etc.
    Each of the components' data structure is managed by their respective handlers and can be accessed individually, if required.

    Args:
        dssfile (str): path of the dss master file (currently only supports OpenDSS files)
        include_DERs (bool, optional): Check whether to include DERs or not. Defaults to True.
        DER_pf (float, optional): Constant power factor of DERs. Defaults to 0.9.
        include_secondary_network (bool, optional): Check whether to include secondary network or not. Defaults to False.

    Examples:
        The only required argument is the OpenDSS master file. We assume that the master file compiles all other OpenDSS files.
        The DSSManager class is initiated first and a method parse_dss() will then parse the overall data.
        >>> dataobj = DSSManager('ieee123master.dss', include_DERs=True)
        >>> dataobj.parse_dss()

    """

    def __init__(
        self,
        dssfile: str,
        include_DERs: bool = False,
        DER_pf: float = 0.9,
        include_secondary_network: bool = False,
    ) -> None:
        """Initialize a DSSManager instance. This instance manages all the components in the distribution system.

        Args:
            dssfile (str): path of the dss master file (currently only supports OpenDSS files)
            include_DERs (bool, optional): Check whether to include DERs or not. Defaults to False.
            DER_pf (float, optional): Constant power factor of DERs. Defaults to 0.9.
            include_secondary_network (bool, optional): Check whether to include secondary network or not. Defaults to False.
        """

        logger.info(f"Initializing DSSManager")

        self.dss = dss
        self.dssfile = dssfile
        # opendss direct checks for filenotfound exception so we do not require any exception here
        self.dss.Text.Command(f"Redirect {self.dssfile}")

        # initialize other attributes
        self.include_DERs = (
            include_DERs  # variable to check whether to include DERs or not
        )
        self.DER_pf = DER_pf  # constant power factor of DERs
        self.include_secondary_network = (
            include_secondary_network  # check whether to include secondary or not
        )

        self.DERs = None  # variable to store information on DERs if included
        self.pv_systems = None

        self.circuit_data = {}  # store circuit metadata such as source bus, base voltage, etc
        # initialize parsing process variables and handlers
        self._initialize()

    @property
    def bus_names(self) -> list[str]:
        """Access all the bus (node) names from the circuit

        Returns:
            list[str]: list of all the bus names
        """
        return self.dss.Circuit.AllBusNames()

    @property
    def basekV_LL(self) -> float:
        """Returns basekV (line to line) of the circuit based on the sourcebus

        Returns:
            float: base kV of the circuit as referred to the source bus
        """
        # make the source bus active before accessing the base kV since there is no provision to get base kV of circuit
        self.dss.Circuit.SetActiveBus(self.source)
        return round(self.dss.Bus.kVBase() * np.sqrt(3), 2)

    @property
    def source(self) -> str:
        """source bus of the circuit.

        Returns:
            str: returns the source bus of the circuit
        """
        # typically the first bus is the source bus
        return self.bus_names[0]

    @timethis
    def _initialize(self) -> None:
        """
        Initialize user-based preferences as well as DSS handlers (i.e. load, transformer, pdelements, and network)
        """
        # if DERs are to be included then include virtual switches for DERs
        if self.include_DERs:
            self._initializeDERs()
            msg_der_initialization = f"DERs virtual switches have been added successfully. The current version assumes a constant power factor of DERs; DERs power factor = {self.DER_pf}"
            logger.info(msg_der_initialization)

            self._initialize_PV()
            msg_pv_initialization = f"DERs virtual switches have been added successfully. The current version assumes a constant power factor of DERs; DERs power factor = {self.DER_pf}"
            logger.info(msg_pv_initialization)

        else:
            logger.info(
                "DERs virtual switches are not included due to exclusion of DERs."
            )

        # initialize DSS handlers
        self._initialize_dsshandlers()

        # initialize data variables (these will be dynamically updated through handlers)
        self.bus_data = None
        self.transformer_data = None
        self.pdelements_data = None
        self.network_graph = None
        self.network_tree = None
        self.normally_open_components = None
        self.load_data = None

    @timethis
    def _initializeDERs(self) -> None:
        """
        Include or exclude virtual switches for DERs based on DER inclusion flag
        """
        self.DERs = []
        generator_flag = self.dss.Generators.First()
        while generator_flag:
            self.dss.Text.Command(
                "New Line.{virtual_DERswitch} phases=3 bus1={source_bus} bus2={gen_bus} switch=True r1=0.001 r0=0.001 x1=0.001 x0=0.001 C1=0 C0=0 length=0.001".format(
                    virtual_DERswitch=self.dss.Generators.Name(),
                    source_bus=self.source,
                    gen_bus=self.dss.Generators.Bus1(),
                )
            )
            self.DERs.append(
                {
                    "name": self.dss.Generators.Name(),
                    "kW_rated": round(self.dss.Generators.kVARated() * self.DER_pf, 2),
                    "connected_bus": self.dss.Generators.Bus1(),
                    "phases": self.dss.Generators.Phases(),
                }
            )

            generator_flag = self.dss.Generators.Next()

        # we also need to ensure that these switches are open as they are virtual switches
        for each_DERs in self.DERs:
            self.dss.Text.Command(f'Open Line.{each_DERs["name"]}')

        self.dss.Solution.Solve()

    @timethis
    def _initialize_PV(self) -> None:
        """
        Include PV Systems from the dss data
        """
        self.pv_systems = []
        pv_flag = self.dss.PVsystems.First()
        while pv_flag:
            self.pv_systems.append(
                {
                    "name": self.dss.PVsystems.Name(),
                    "kW_rated": round(
                        self.dss.PVsystems.kVARated() * self.dss.PVsystems.pf(), 2
                    ),
                    "connected_bus": self.dss.CktElement.BusNames()[0],
                    "phases": self.dss.CktElement.NumPhases(),
                }
            )

            pv_flag = self.dss.PVsystems.Next()

    @timethis
    def _initialize_dsshandlers(self) -> None:
        """Initialize all the DSS Handlers"""

        # bus_handler is currently not being used here but kept here for future usage
        self.bus_handler = BusHandler(self.dss)
        self.transformer_handler = TransformerHandler(self.dss)
        self.pdelement_handler = PDElementHandler(self.dss)
        self.network_handler = NetworkHandler(
            self.dss, pdelement_handler=self.pdelement_handler
        )

        if self.include_secondary_network:
            logger.info("Considering entire system including secondary networks")
            self.load_handler = LoadHandler(
                self.dss, include_secondary_network=self.include_secondary_network
            )
        else:
            # if primary loads are to be referred then we must pass network and transformer handlers
            logger.info(
                "Considering primary networks and aggregating loads by referring them to the primary node"
            )
            self.load_handler = LoadHandler(
                self.dss,
                include_secondary_network=self.include_secondary_network,
                network_handler=self.network_handler,
                transformer_handler=self.transformer_handler,
            )
        logger.info(
            f'Successfully instantiated required handlers from "{self.dssfile}"'
        )

    @timethis
    def parsedss(self) -> None:
        """Parse required data from the handlers to respective class variables"""
        self.bus_data = self.bus_handler.get_buses()
        self.transformer_data = self.transformer_handler.get_transformers()
        self.pdelements_data = self.pdelement_handler.get_pdelements()
        self.network_graph, self.network_tree, self.normally_open_components = (
            self.network_handler.network_topology()
        )
        self.load_data = self.load_handler.get_loads()

        if not self.include_secondary_network:
            logger.info(
                f"Excluding secondaries from final tree, graph configurations, and pdelements."
            )
            self.network_tree.remove_nodes_from(
                self.load_handler.downstream_nodes_from_primary
            )
            self.network_graph.remove_nodes_from(
                self.load_handler.downstream_nodes_from_primary
            )
            self.pdelements_data = [
                items
                for items in self.pdelements_data
                if items["from_bus"]
                not in self.load_handler.downstream_nodes_from_primary
                and items["to_bus"]
                not in self.load_handler.downstream_nodes_from_primary
            ]
        logger.info(f"Successfully parsed the required data from all handlers.")

        # the networkx data is saved as a serialized JSON
        self.network_graph_data = json_graph.node_link_data(self.network_graph)
        self.network_tree_data = json_graph.node_link_data(self.network_tree)

        # parse additional circuit data
        # add more as required in the future ...
        self.circuit_data = {
            "substation": self.source,
            "basekV_LL_circuit": self.basekV_LL,
        }

    @timethis
    def saveparseddss(
        self, folder_name: str = f"parsed_data", folder_exist_ok: bool = False
    ) -> None:
        """Saves the parsed data from all the handlers

        Args:
            folder_name (str, optional): Name of the folder to save the data in. Defaults to "dssdatatocsv"_<current system date>.
            folder_exist_ok (bool, optional): Boolean to check if folder rewrite is ok. Defaults to False.
        """

        # check if parsedss is run before saving these files
        if self.bus_data is None:
            logger.error(
                "Please run DSSManager.parsedss() to parse the data and then run this function to save the files."
            )
            raise NotImplementedError(
                f"Data variables are empty. You must run {__name__}.DSSManager.parsedss() to extract the data before saving them."
            )

        # check if the path already exists. This prevent overwrite
        try:
            Path(folder_name).mkdir(parents=True, exist_ok=folder_exist_ok)
        except FileExistsError:
            logger.error(
                "The folder already exists and the module is attempting to rewrite the data in the folder. Either provide a path in <folder_name> or mention <folder_exist_ok=True> to rewrite the existing files."
            )
            raise FileExistsError(
                "The folder or files already exist. Please provide a non-existent path."
            )

        # save all the data in the new folder
        # the non-networkx data are all saved as dataframe in csv
        pd.DataFrame(self.bus_data).to_csv(f"{folder_name}/bus_data.csv", index=False)
        pd.DataFrame(self.transformer_data).to_csv(
            f"{folder_name}/transformer_data.csv", index=False
        )
        pd.DataFrame(self.pdelements_data).to_csv(
            f"{folder_name}/pdelements_data.csv", index=False
        )
        pd.DataFrame(self.load_data).to_csv(f"{folder_name}/load_data.csv", index=False)
        pd.DataFrame(
            self.normally_open_components, columns=["normally_open_components"]
        ).to_csv(f"{folder_name}/normally_open_components.csv", index=False)

        if self.DERs is not None:
            pd.DataFrame(self.DERs).to_csv(f"{folder_name}/DERs.csv", index=False)
            pd.DataFrame(self.pv_systems).to_csv(
                f"{folder_name}/pv_systems.csv", index=False
            )

        with open(f"{folder_name}/network_graph_data.json", "w") as file:
            json.dump(self.network_graph_data, file)

        with open(f"{folder_name}/network_tree_data.json", "w") as file:
            json.dump(self.network_tree_data, file)

        # save the circuit data as json
        with open(f"{folder_name}/circuit_data.json", "w") as file:
            json.dump(self.circuit_data, file)

        logger.info("Successfully saved required files.")

---

BusHandler

ldrestoration.dssparser.bushandler.BusHandler

BusHandler deals with bus (node) related data from the distribution model.

Parameters:

Name	Type	Description	Default
dss_instance	ModuleType	redirected opendssdirect instance	required

Note

Bus and Nodes are two different concepts in distribution systems modeling and are used interchangably here for simplicity.

Source code in ldrestoration/dssparser/bushandler.py

class BusHandler:
    """BusHandler deals with bus (node) related data from the distribution model.

    Args:
        dss_instance (ModuleType): redirected opendssdirect instance 

    Note: 
        Bus and Nodes are two different concepts in distribution systems modeling and are used interchangably here 
        for simplicity.    
    """

    def __init__(self, 
                 dss_instance: ModuleType) -> None:
        """Initialize a BusHandler instance. This instance deals with bus (node) related data from the distribution model.
        Note: Bus and Nodes are two different concepts in distribution systems modeling and are used interchangably here 
        for simplicity.

        Args:
            dss_instance (ModuleType): redirected opendssdirect instance 
        """

        self.dss_instance = dss_instance 

    @timethis
    def get_buses(self) -> list[dict[str,Union[int,str,float]]]: 
        """Extract the bus data -> name, basekV, latitude, longitude from the distribution model.

        Returns:
            bus_data (list[dict[str,Union[int,str,float]]]): list of bus data for each buses
        """
        all_buses_names = self.dss_instance.Circuit.AllBusNames()        
        bus_data = []        
        for bus in all_buses_names:

            # need to set the nodes active before extracting their info 
            self.dss_instance.Circuit.SetActiveBus(bus)

            # be careful that X gives you lon and Y gives you lat
            bus_coordinates = dict(name = self.dss_instance.Bus.Name(),
                                   basekV = round(self.dss_instance.Bus.kVBase(),2),
                                   latitude = self.dss_instance.Bus.Y(),
                                   longitude = self.dss_instance.Bus.X())

            bus_data.append(bus_coordinates)
        return bus_data

---

LoadHandler

ldrestoration.dssparser.loadhandler.LoadHandler

LoadHandler deals with all the loads in the distribution system. When include_secondary_network=False, all the secondary loads are referred back to their primary.

Parameters:

Name	Type	Description	Default
dss_instance	ModuleType	redirected opendssdirect instance	required
network_handler	Optional[NetworkHandler]	Directed network tree of the distribution model, Defaults to None	None
transformer_handler	Optional[TransformerHandler]	Instance of TransformerHandler. Defaults to None.	None
include_secondary_network	Optional[bool]	Whether the secondary network is to be considered or not, Defaults to False	False
bus_names	Optional[list[str]]	Names of all the buses (nodes) in the distribution model	None

Note

• In OpenDSS, the phases information of loads are lost on the secondary side of the split-phase transformers. Hence, each of the loads are traced back to their nearest transformer to identify the corresponding phase. For delta primary, the loads are equally distributed to each phase.

To do

• The current version does not address phase wise load decoupling for delta connected loads. It will be incorporated in the future releases.

Source code in ldrestoration/dssparser/loadhandler.py

class LoadHandler:
    """LoadHandler deals with all the loads in the distribution system. 
    When `include_secondary_network=False`, all the secondary loads are referred back to their primary.  

    Args:
        dss_instance (ModuleType): redirected opendssdirect instance
        network_handler (Optional[NetworkHandler]): Directed network tree of the distribution model, Defaults to None
        transformer_handler (Optional[TransformerHandler]): Instance of TransformerHandler. Defaults to None.
        include_secondary_network (Optional[bool]): Whether the secondary network is to be considered or not, Defaults to False
        bus_names (Optional[list[str]]):Names of all the buses (nodes) in the distribution model

    Note:
        * In OpenDSS, the phases information of loads are lost on the secondary side of the split-phase transformers. Hence, each of the loads are traced back to their 
        nearest transformer to identify the corresponding phase. For delta primary, the loads are equally distributed to each phase. 

    To do:
        * The current version does not address phase wise load decoupling for delta connected loads. It will be incorporated in the future releases.

    """     

    def __init__(self, 
                 dss_instance: ModuleType,
                 network_handler: Optional[NetworkHandler] = None,
                 transformer_handler: Optional[TransformerHandler] = None, 
                 include_secondary_network: Optional[bool] = False,                 
                 bus_names: Optional[list[str]] = None) -> None:

        """Initialize a LoadHandler instance. This instance deals with all the loads in the distribution system. 

        Args:
            dss_instance (ModuleType): redirected opendssdirect instance
            network_handler (Optional[NetworkHandler]): Directed network tree of the distribution model, Defaults to None
            transformer_handler (Optional[TransformerHandler]): Instance of TransformerHandler. Defaults to None.
            include_secondary_network (Optional[bool]): Whether the secondary network is to be considered or not, Defaults to False
            bus_names (Optional[list[str]]):Names of all the buses (nodes) in the distribution model
        """        

        self.dss_instance = dss_instance         
        self.network_handler = network_handler
        self.transformer_handler = transformer_handler
        self.include_secondary_network = include_secondary_network

        # since bus_names is required for any methods in LoadHandler, we rather check it in the initialization
        self.bus_names = self.dss_instance.Circuit.AllBusNames() if bus_names is None else bus_names        
        self.downstream_nodes_from_primary = None

        # validate if the required inputs are in existence
        self.__load_input_validator()

    @timethis
    def __load_input_validator(self) -> None:
            """This is to be checked in the future. Network and Transformer handler should be optional
            and only available if loads are to be referred to the primary."""

            if not self.include_secondary_network and (not self.transformer_handler and not self.network_handler):
                # if we do not want secondary and we do not pass any handlers then there must be an error
                logger.warning("You need to provide NetworkHandler() and TransformerHandler as arguments to LoadHandler()")
                raise NotImplementedError(
                "To refer the loads to primary, both NetworkHandler and TransformerHandler are required."
                )  

    @timethis
    def get_loads(self) -> pd.DataFrame:

        if self.include_secondary_network:
            # get all loads as they appear in the secondary
            logger.info("Fetching the loads as they appear on the secondary")
            return self.get_all_loads()
        else:
            # get primarry referred loads
            logger.info("Referring the loads back to the primary node of the distribution transformer.")
            return self.get_primary_referred_loads()             

    @property
    @cache
    def bus_names_to_index_map(self) -> dict[str,int]:
        """each of the bus mapped to its corresponding index in the bus names list

        Returns:
            dict[str,int]: dictionary with key as bus names and value as its index
        """        
        return {bus:index for index,bus in enumerate(self.bus_names)}

    @timethis    
    def get_all_loads(self) -> pd.DataFrame:
        """Extract load information for each bus(node) for each phase. This method extracts load on the exact bus(node) as 
        modeled in the distribution model, including secondary.

        Returns:
            load_per_phase(pd.DataFrame): Per phase load data in a pandas dataframe
        """

        num_buses = len(self.bus_names)

        # Initialize arrays for load_per_phase
        load_per_phase = {
            "name": [""] * num_buses,
            "bus": self.bus_names,
            "P1": np.zeros(num_buses),
            "Q1": np.zeros(num_buses),
            "P2": np.zeros(num_buses),
            "Q2": np.zeros(num_buses),
            "P3": np.zeros(num_buses),
            "Q3": np.zeros(num_buses)
        }

        loads_flag = self.dss_instance.Loads.First()

        while loads_flag:
            connected_buses = self.dss_instance.CktElement.BusNames()           

            # conductor power contains info on active and reactive power
            conductor_power = np.array(self.dss_instance.CktElement.Powers())
            nonzero_power_indices = np.where(conductor_power != 0)[0]
            nonzero_power = conductor_power[nonzero_power_indices]

            for buses in connected_buses:
                bus_split = buses.split(".")
                if (len(bus_split) == 4) or (len(bus_split) == 1):

                    # three phase checker
                    connected_bus = bus_split[0]
                    bus_index = self.bus_names_to_index_map[connected_bus]
                    load_per_phase["name"][bus_index] = self.dss_instance.Loads.Name()
                    P_values = nonzero_power[::2]   # Extract P values (every other element starting from the first)
                    Q_values = nonzero_power[1::2]  # Extract Q values (every other element starting from the second)
                    for phase_index in range(3):
                        load_per_phase[f"P{phase_index + 1}"][bus_index] += round(P_values[phase_index],2)
                        load_per_phase[f"Q{phase_index + 1}"][bus_index] += round(Q_values[phase_index],2)

                else:
                    # non three phase load
                    connected_bus, connected_phase_secondary = bus_split[0], bus_split[1:]
                    bus_index = self.bus_names_to_index_map[connected_bus]                  
                    load_per_phase["name"][bus_index] = self.dss_instance.Loads.Name()
                    P_values = nonzero_power[::2]  # Extract P values (every alternate element starting from the first)
                    Q_values = nonzero_power[1::2]  # Extract Q values (every alternate element starting from the second)

                    for phase_index, phase in enumerate(connected_phase_secondary):
                        load_per_phase[f"P{phase}"][bus_index] += round(P_values[phase_index], 2)
                        load_per_phase[f"Q{phase}"][bus_index] += round(Q_values[phase_index], 2)

            loads_flag = self.dss_instance.Loads.Next()

        return pd.DataFrame(load_per_phase)        

    @timethis
    def get_primary_referred_loads(self) -> pd.DataFrame:
        """Transfer all the secondary nodes to the primary corresponding to each split phase transformer. 
        Also returns the downstream nodes from the split phase transformers.

        Returns:
            primary_loads_df(pd.DataFrame): Per phase load data in a pandas dataframe with secondary transferred to primary
        """
        # keep track of all downstream nodes from primary. This is for removal from network as well,
        # since we are aggregating these loads in the primary, removing them will reduce computational burden
        self.downstream_nodes_from_primary = set()  

        # get access to the network tree from the topology
        _, network_tree, _ = self.network_handler.network_topology()

        # obtain the relation between the primary phase and secondary bus in splitphase transformer
        # this obtains a dictionary with secondary nodes s key and their associated phase info as value
        split_phase_primary = self.transformer_handler.get_splitphase_primary()

        # initially this is the secondary load but will be changed to reflect the primary load referral
        primary_loads_df = self.get_all_loads()   

        for xfrmr_secondary_node, primary_phase in split_phase_primary.items():

            # here we get the predecessor of the secondary i.e. primary node
            xfrmr_primary_node = list(network_tree.predecessors(xfrmr_secondary_node))[0]

            # identify the secondary and primary bus indices so that loads are referred to primary             
            secondary_bus_index = self.bus_names_to_index_map[xfrmr_secondary_node]
            primary_bus_index = self.bus_names_to_index_map[xfrmr_primary_node]

            # however we still traverse downstream from the secondary as traversing from primary could follow other routes too
            xfrmr_downstream_nodes = nx.descendants(network_tree, xfrmr_secondary_node)
            xfrmr_downstream_nodes.add(xfrmr_secondary_node)

            # add all the nodes downstream from the transformer's primary (including secondary)
            self.downstream_nodes_from_primary.update(xfrmr_downstream_nodes)      

            # now we refer all the downstream node loads (if available) to the primary and remove them from the load df
            # this reduces computational burden when not dealing with the secondaries
            for xfrmr_downstream_node in xfrmr_downstream_nodes:   
                try:
                    downstream_node_index = self.bus_names_to_index_map[xfrmr_downstream_node]
                except KeyError:
                    logger.error(f"Invalid load name {xfrmr_downstream_node}")
                    raise KeyError(f"{xfrmr_downstream_node} is not a valid node name.")

                primary_loads_df.loc[primary_bus_index, f"P{primary_phase[0]}"] += (primary_loads_df["P1"][downstream_node_index] +
                                                                                    primary_loads_df["P2"][downstream_node_index]) 
                primary_loads_df.loc[primary_bus_index, f"Q{primary_phase[0]}"] += (primary_loads_df["Q1"][downstream_node_index] +
                                                                                    primary_loads_df["Q2"][downstream_node_index])                

                primary_loads_df.loc[primary_bus_index, f"name"] = primary_loads_df["name"][downstream_node_index]

                # drop the secondaries from the dataframe
                primary_loads_df.drop(downstream_node_index, inplace=True) 
                # primary_loads_df.drop(secondary_bus_index, inplace=True)  

        # reset the loads dataframe to its original index
        primary_loads_df.reset_index(inplace=True, drop=True)

        return primary_loads_df 

---

NetworkHandler

ldrestoration.dssparser.networkhandler.NetworkHandler

NetworkHandler creates and modifies the network as a graph (nodes and edges) for the distribution model

Parameters:

Name	Type	Description	Default
dss_instance	ModuleType	redirected opendssdirect instance	required
bus_names	Optional[list[str]]	Names of all the buses (nodes) in the distribution model. Defaults to None	None
source	Optional[str]	Source node of the graph to build. Defaults to None.	None
pdelement_handler	Optional[PDElementHandler]	Instance of PDElementHandler. Defaults to None.	None
pdelements_data	Optional[list[dict[str, Union[int, str, float]]]]	All the required data of the pdelements(edges) from PDElementHandler or provided by user in pdelements format. Defaults to None.	None

To do

• extract feeder information to visualize it better
• add a method to remove list of edges provided as an argument

Source code in ldrestoration/dssparser/networkhandler.py

class NetworkHandler:

    """NetworkHandler creates and modifies the network as a graph (nodes and edges) for the distribution model 

    Args:
        dss_instance (ModuleType): redirected opendssdirect instance
        bus_names (Optional[list[str]]):Names of all the buses (nodes) in the distribution model. Defaults to None
        source (Optional[str], optional): Source node of the graph to build. Defaults to None.
        pdelement_handler (Optional[PDElementHandler], optional): Instance of PDElementHandler. Defaults to None.
        pdelements_data (Optional[list[dict[str,Union[int,str,float]]]], optional): All the required data of the pdelements(edges) from PDElementHandler or provided by user in pdelements format. Defaults to None.

    To do:
        * extract feeder information to visualize it better
        * add a method to remove list of edges provided as an argument
    """   
    def __init__(self, 
                 dss_instance: ModuleType, 
                 bus_names: Optional[list[str]] = None,
                 source: Optional[str] = None,
                 pdelement_handler: Optional[PDElementHandler] = None,
                 pdelements_data: Optional[list[dict[str,Union[int,str,float, npt.NDArray[np.complex128]]]]] = None) -> None:

        """Initialize a NetworkHandler instance. Create and modify the network as a graph (nodes and edges) for the distribution model 

        Args:
            dss_instance (ModuleType): redirected opendssdirect instance
            bus_names (Optional[list[str]]):Names of all the buses (nodes) in the distribution model. Defaults to None
            source (Optional[str], optional): Source node of the graph to build. Defaults to None.
            pdelement_handler (Optional[PDElementHandler], optional): Instance of PDElementHandler. Defaults to None.
            pdelements_data (Optional[list[dict[str,Union[int,str,float]]]], optional): All the required data of the pdelements(edges) 
            from PDElementHandler or provided by user in pdelements format. Defaults to None.
        """                      

        self.dss_instance = dss_instance 
        self.bus_names = bus_names
        self.source = source        
        self.pdelements_data = pdelements_data
        self.pdelement_handler = pdelement_handler

    def __network_input_validator(self) -> None:
        """Validates the data required to build a network

        Raises:
            NotImplementedError (elements): This is raised when pdelements data are not provided (either json or PDElementHandler)
            EOFError (source): This is raised when a source does not exist in the tree i.e. the bus of the distribution model
        """        
        if self.bus_names is None:
            logger.info("Bus names not provided. So extracting it from the base network.")
            self.bus_names = self.dss_instance.Circuit.AllBusNames()

        if self.pdelements_data is None and self.pdelement_handler is None:
            logger.warning("You need to provide either one of the following: pcelement list of dicts or PCElementHandler")
            raise NotImplementedError(
                "Please provide an element file (json) OR PDElement instance from PDElement handler to create the network."
                )

        if self.source is not None and self.source not in self.bus_names:
            logger.warning("The source must be one of the existing buses (nodes) in the distribution model.")
            raise EOFError(
                "Please provide a valid source. A source must be an existing bus (node) in the distribution model"
                )

    def __set_node_coordinates(self, network_tree: nx.DiGraph) -> None:
        """Sets the coordinates of each nodes as per the bus coordinates data

        Args:
            network_tree (nx.DiGraph): A directed graph (network tree in this case)
        """

        for node in network_tree.nodes():

            # need to set the nodes active before extracting their info 
            self.dss_instance.Circuit.SetActiveBus(node)

            # be careful that X gives you lon and Y gives you lat
            network_tree.nodes[node]['lat'] = self.dss_instance.Bus.Y()
            network_tree.nodes[node]['lon'] = self.dss_instance.Bus.X()


    def build_network(self, all_pdelements: list[dict[str,Union[int,str,float, npt.NDArray[np.complex128]]]]) -> tuple[nx.Graph, list[str]]: 
        """Build the network from the pdelements data

        Args:
            all_pdelements (list[dict[str,Union[int,str,float, npt.NDArray[np.complex128]]]]): All the required data of the pdelements(edges) 
            from PDElementHandler or provided by user in pdelements format.

        Returns:
            network_graph (nx.Graph): Network graph (undirected) 
            normally_open_components (list[str]): Names of normally open pdelements (tie or virtual switches) 
        """         

        # initiate a network graph. Since the power flow occurs in any direction this should be undirected
        network_graph = nx.Graph()

        # extract normally open components
        normally_open_components = []

        # add lines as graph edges. initial version may have loops so we create graphs and then create trees
        for each_line in all_pdelements:
            if not each_line['is_open']:
                network_graph.add_edge(each_line['from_bus'],
                                       each_line['to_bus'], 
                                       # the remaining arguments are the data associated with each edges
                                       element=each_line['element'],
                                       is_switch=each_line['is_switch'],
                                       is_open=each_line['is_open'],
                                       name=each_line['name'])
            else:
                normally_open_components.append(each_line['name'])

        return network_graph, normally_open_components          

    def network_topology(self) -> tuple[nx.Graph, nx.DiGraph, list[str]]:
        """Create network topology including graph, trees, and open components

        Returns:
            network_graph (nx.Graph): Undirected network graph
            network_tree (list[str]): Directed network tree
            normally_open_components (list[str]): Names of normally open pdelements (tie or virtual switches)
        """

        # validate the data first i.e. whether pdelements were provided or not
        self.__network_input_validator()
        self.source = self.bus_names[0]

        # this can be user defined lines or can be extracted from the base network using PDElementHandler
        all_pdelements = self.pdelement_handler.get_pdelements() if not self.pdelements_data else self.pdelements_data

        network_graph, normally_open_components = self.build_network(all_pdelements)
        network_tree = nx.bfs_tree(network_graph, source=self.source)

        # add the bus corrdinates from dss to network tree
        self.__set_node_coordinates(network_tree)   

        return network_graph, network_tree, normally_open_components

---

PDElementHandler

ldrestoration.dssparser.pdelementshandler.PDElementHandler

PDElementHandler deals with all the power delivery elements -> lines, transformers, reactors, and capacitors. ALthough we have separate handlers for a few of them, we extract the PDelements here as they represent edges for out network

Parameters:

Name	Type	Description	Default
dss_instance	ModuleType	redirected opendssdirect instance	required

Source code in ldrestoration/dssparser/pdelementshandler.py

class PDElementHandler:
    """PDElementHandler deals with all the power delivery elements -> lines, transformers,
    reactors, and capacitors. ALthough we have separate handlers for a few of them, we extract the PDelements here as they represent
    edges for out network

    Args:
        dss_instance (ModuleType): redirected opendssdirect instance
    """

    def __init__(self, dss_instance: ModuleType) -> None:
        """Initialize a PDElementHandler instance. This instance deals with all the power delivery elements -> lines, transformers,
        reactors, and capacitors. ALthough we have separate handlers for a few of them, we extract the PDelements here as they represent
        edges for out network

        Args:
            dss_instance (ModuleType): redirected opendssdirect instance
        """

        self.dss_instance = dss_instance

    def __get_line_zmatrix(self) -> tuple[np.ndarray, np.ndarray]:
        """Returns the z_matrix of a specified line element.

        Returns:
            real z_matrix, imag z_matrix (np.ndarray, np.ndarray): 3x3 numpy array of the z_matrix corresponding to the each of the phases(real,imag)
        """

        if (len(self.dss_instance.CktElement.BusNames()[0].split(".")) == 4) or (
            len(self.dss_instance.CktElement.BusNames()[0].split(".")) == 1
        ):

            # this is the condition check for three phase since three phase is either represented by bus_name.1.2.3 or bus_name
            z_matrix = np.array(self.dss_instance.Lines.RMatrix()) + 1j * np.array(
                self.dss_instance.Lines.XMatrix()
            )
            z_matrix = z_matrix.reshape(3, 3)

            return np.real(z_matrix), np.imag(z_matrix)

        else:

            # for other than 3 phases
            active_phases = [
                int(phase)
                for phase in self.dss_instance.CktElement.BusNames()[0].split(".")[1:]
            ]
            z_matrix = np.zeros((3, 3), dtype=complex)
            r_matrix = self.dss_instance.Lines.RMatrix()
            x_matrix = self.dss_instance.Lines.XMatrix()
            counter = 0
            for _, row in enumerate(active_phases):
                for _, col in enumerate(active_phases):
                    z_matrix[row - 1, col - 1] = complex(
                        r_matrix[counter], x_matrix[counter]
                    )
                    counter = counter + 1

            return np.real(z_matrix), np.imag(z_matrix)

    def __get_nonline_zmatrix(self) -> list[list[float]]:
        """Returns the z_matrix of a specified element other than the line element.

        Returns:
            z_matrix (list[list[float]]): list of list of float of z matrices (same fo real and imag)
        """
        # hash map for the element z matrices other than lines
        # this is temporary and should be fixed later to replace with the actual impedances of the element.
        elements_z_matrix = {
            ("1",): [[0.001, 0, 0], [0, 0, 0], [0, 0, 0]],
            ("2",): [[0, 0, 0], [0, 0.001, 0], [0, 0, 0]],
            ("3",): [[0, 0, 0], [0, 0, 0], [0, 0, 0.001]],
            ("1", "2"): [[0.001, 0, 0], [0, 0.001, 0], [0, 0, 0]],
            ("2", "3"): [[0, 0, 0], [0, 0.001, 0], [0, 0, 0.001]],
            ("1", "3"): [[0.001, 0, 0], [0, 0, 0], [0, 0, 0.001]],
            ("1", "2", "3"): [[0.001, 0, 0], [0, 0.001, 0], [0, 0, 0.001]],
        }

        if self.dss_instance.CktElement.NumPhases() == 3:
            return elements_z_matrix[("1", "2", "3")]
        else:
            bus_phases = self.dss_instance.CktElement.BusNames()[0].split(".")[1:]
            return elements_z_matrix[tuple(bus_phases)]

    def element_phase_identification(self, element_phases=list[str]) -> list[str]:
        """Match the phase from the number convention to letter convention i.e. 123 -> abc

        Returns:
            set[str]: set of phases converted to letter type
        """
        # create a dict of mapper
        phasemap = {"1": "a", "2": "b", "3": "c"}

        # identify and return corresponding dss phases in numbers to the letters
        return {phasemap[dss_phase] for dss_phase in element_phases}

    @cached_property
    def transformer_rating(self) -> int:
        """Obtain transformer rating for each transformer

        Returns:
            int: Rating of the transformer in kVA (assume this to be kW for now)
        """

        each_transformer_rating = {}
        flag = self.dss_instance.Transformers.First()
        while flag:
            transformer_name = self.dss_instance.Transformers.Name()
            each_transformer_rating[transformer_name] = (
                self.dss_instance.Transformers.kVA()
            )
            flag = self.dss_instance.Transformers.Next()

        return each_transformer_rating

    def get_pdelements(self) -> list[dict[str, Union[int, str, float, np.ndarray]]]:
        """Extract the list of PDElement from the distribution model. Capacitors are excluded.

        Returns:
            pdelement_list (list[dict[str,Union[int,str,float, np.ndarray]]]):
            list of pdelements with required information
        """

        element_activity_status = self.dss_instance.PDElements.First()
        pdelement_list = []

        while element_activity_status:
            element_type = self.dss_instance.CktElement.Name().lower().split(".")[0]

            # capacitor is a shunt element  and is not included
            if element_type != "capacitor":
                # "Capacitors are shunt elements and are not modeled in this work. Regulators are not modeled as well."
                if element_type == "line":
                    z_matrix_real, z_matrix_imag = self.__get_line_zmatrix()
                    each_element_data = {
                        "name": self.dss_instance.Lines.Name(),
                        "element": element_type,
                        # from opendss manual -> length units = {none|mi|kft|km|m|ft|in|cm}
                        "length_unit": self.dss_instance.Lines.Units(),
                        "z_matrix_real": z_matrix_real.tolist(),
                        "z_matrix_imag": z_matrix_imag.tolist(),
                        "length": self.dss_instance.Lines.Length(),
                        "from_bus": self.dss_instance.Lines.Bus1().split(".")[0],
                        "to_bus": self.dss_instance.Lines.Bus2().split(".")[0],
                        "num_phases": self.dss_instance.Lines.Phases(),
                        "phases": (
                            {"a", "b", "c"}
                            if self.dss_instance.CktElement.NumPhases() == 3
                            else self.element_phase_identification(
                                element_phases=self.dss_instance.CktElement.BusNames()[
                                    0
                                ].split(".")[1:]
                            )
                        ),
                        "is_switch": self.dss_instance.Lines.IsSwitch(),
                        "is_open": (
                            self.dss_instance.CktElement.IsOpen(1, 0)
                            or self.dss_instance.CktElement.IsOpen(2, 0)
                        ),
                    }
                    # obtain the kVbase (line to line) of the element
                    # we assume the voltage level of the element is the voltage of its secondary bus

                    self.dss_instance.Circuit.SetActiveBus(
                        self.dss_instance.Lines.Bus2().split(".")[0]
                    )
                    each_element_data["base_kv_LL"] = round(
                        self.dss_instance.Bus.kVBase() * np.sqrt(3), 2
                    )

                    # the loading is per conductor
                    each_element_data["normal_loading_kW"] = (
                        each_element_data["base_kv_LL"]
                        / np.sqrt(3)
                        * self.dss_instance.Lines.NormAmps()
                        if each_element_data["num_phases"] == 1
                        else each_element_data["base_kv_LL"]
                        * np.sqrt(3)
                        * self.dss_instance.Lines.NormAmps()
                    )

                    each_element_data["emergency_loading_kW"] = (
                        each_element_data["base_kv_LL"]
                        / np.sqrt(3)
                        * self.dss_instance.Lines.EmergAmps()
                        if each_element_data["num_phases"] == 1
                        else each_element_data["base_kv_LL"]
                        * np.sqrt(3)
                        * self.dss_instance.Lines.EmergAmps()
                    )

                else:
                    # everything other than lines but not capacitors i.e. transformers, reactors etc.
                    # The impedance matrix for transformers and reactors are modeled as a shorted line here.
                    # Need to work on this for future cases and replace with their zero sequence impedance may be

                    each_element_data = {
                        "name": self.dss_instance.CktElement.Name().split(".")[1],
                        "element": element_type,
                        # from opendss manual -> length units = {none|mi|kft|km|m|ft|in|cm}
                        "length_unit": 0,
                        "z_matrix_real": self.__get_nonline_zmatrix(),
                        "z_matrix_imag": self.__get_nonline_zmatrix(),
                        "length": 0.001,
                        "from_bus": self.dss_instance.CktElement.BusNames()[0].split(
                            "."
                        )[0],
                        "to_bus": self.dss_instance.CktElement.BusNames()[1].split(".")[
                            0
                        ],
                        # for non lines dss.Lines does not work so we need to work around with CktElement
                        # CktElement is activated along with PDElements
                        "num_phases": self.dss_instance.CktElement.NumPhases(),
                        "phases": (
                            {"a", "b", "c"}
                            if self.dss_instance.CktElement.NumPhases() == 3
                            else self.element_phase_identification(
                                element_phases=self.dss_instance.CktElement.BusNames()[
                                    0
                                ].split(".")[1:]
                            )
                        ),
                        "is_switch": False,
                        "is_open": False,
                    }

                    # obtain the kVbase (line to line) of the element
                    # we assume the voltage level of the element is the voltage of its secondary bus
                    self.dss_instance.Circuit.SetActiveBus(
                        self.dss_instance.CktElement.BusNames()[1].split(".")[0]
                    )
                    each_element_data["base_kv_LL"] = round(
                        self.dss_instance.Bus.kVBase() * np.sqrt(3), 2
                    )

                    # loading on the transformer is also on per phase basis
                    if element_type == "transformer":
                        each_element_data["normal_loading_kW"] = (
                            self.transformer_rating[each_element_data["name"]]
                            / each_element_data["num_phases"]
                        )

                        # setting emergency loading to 150% of the normal loading
                        each_element_data["emergency_loading_kW"] = (
                            1.5 * each_element_data["normal_loading_kW"]
                        )

                    else:
                        self.dss_instance.Transformers.First()
                        rating_substation = self.dss_instance.Transformers.kVA()
                        each_element_data["normal_loading_kW"] = (
                            rating_substation / each_element_data["num_phases"]
                        )

                        # setting emergency loading to 150% of the normal loading
                        each_element_data["emergency_loading_kW"] = (
                            1.5 * each_element_data["normal_loading_kW"]
                        )

                pdelement_list.append(each_element_data)
            element_activity_status = self.dss_instance.PDElements.Next()

        return pdelement_list

---

TransformerHandler

ldrestoration.dssparser.transformerhandler.TransformerHandler

TransformerHandler extracts the transformers (step down, step up, or split-phase service transformers) in the distribution model.

Parameters:

Name	Type	Description	Default
dss_instance	ModuleType	redirected opendssdirect instance	required

To do

• Address the extraction of delta connected primary in split-phase transformers

Source code in ldrestoration/dssparser/transformerhandler.py

class TransformerHandler:
    """TransformerHandler extracts the transformers (step down, step up, or split-phase service transformers) in the distribution model.

    Args:
        dss_instance (ModuleType): redirected opendssdirect instance 

    To do:
        * Address the extraction of delta connected primary in split-phase transformers
    """
    def __init__(self, 
                 dss_instance: ModuleType) -> None:
        """Initialize a TransformerHandler instance. This instance deals with transformers in the distribution model.

        Args:
            dss_instance (ModuleType): redirected opendssdirect instance 
        """

        self.dss_instance = dss_instance 

    def get_splitphase_primary(self) -> dict[str,str]:    
        """Gets the primary phase information from split phase transformers to refer all loads to the primary

        Returns:
            splitphase_node_primary (dict[str,str]): A dictionary with secondary node as key and associated phase in primary as value
            for eg. for ['A.3', 'B.1.0', 'B.0.2'] this will return {'B':['3']}
        """   
        splitphase_node_primary = {}                
        transformer_flag = self.dss_instance.Transformers.First()        
        while transformer_flag:

            if (self.dss_instance.CktElement.NumPhases() != 3) and self.dss_instance.Transformers.NumWindings() == 3:
                # a split phase transformer is a three winding single phase transformer (two phase primary accounts for delta)

                # name extracted of the secondary and phase extracted of the primary
                bus_name = self.dss_instance.CktElement.BusNames()[1].split('.')[0]
                bus_phases = self.dss_instance.CktElement.BusNames()[0].split('.')[1:]

                if bus_name not in splitphase_node_primary:
                    splitphase_node_primary[bus_name] = bus_phases

            transformer_flag = self.dss_instance.Transformers.Next()

        return splitphase_node_primary

    def get_transformers(self) -> list[dict[str,Union[int,str,float]]]: 
        """Extract the bus data -> name, basekV, latitude, longitude from the distribution model.

        Returns:
            bus_data (list[dict[str,Union[int,str,float]]]): list of bus data for each buses
        """
        transformer_flag = self.dss_instance.Transformers.First()      
        transformer_data = []  
        while transformer_flag:
            each_transformer = dict(name = self.dss_instance.Transformers.Name(),
                                    numwindings = self.dss_instance.Transformers.NumWindings(),
                                    connected_from = self.dss_instance.CktElement.BusNames()[0].split('.')[0],
                                    connected_to = self.dss_instance.CktElement.BusNames()[1].split('.')[0])

            transformer_data.append(each_transformer)
            transformer_flag = self.dss_instance.Transformers.Next() 
        return transformer_data