refactor: move common grid code to AbstractGrid

powersimdata/input/abstract_grid.py

@@ -1,15 +1,25 @@
+import os
+
 import pandas as pd
 
 from powersimdata.input import const
+from powersimdata.input.helpers import (
+    add_coord_to_grid_data_frames,
+    add_zone_to_grid_data_frames,
+    csv_to_data_frame,
+)
+from powersimdata.network.csv_reader import CSVReader
 
 
 class AbstractGrid:
-    """Grid Builder."""
+    """Grid Builder. Child classes must assign self.top_dirname and
+    self.umbrella_interconnect before self.__init__ is called, or re-define the __init__
+    and/or methods called within the __init__ to avoid an AttributeError.
+    """
 
     def __init__(self):
         """Constructor"""
         self.data_loc = None
-        self.interconnect = None
         self.zone2id = {}
         self.id2zone = {}
         self.sub = pd.DataFrame()
@@ -20,6 +30,63 @@ def __init__(self):
         self.bus = pd.DataFrame()
         self.branch = pd.DataFrame()
         self.storage = storage_template()
+        self._set_data_loc()
+        self._build_network()
+
+    def _set_data_loc(self):
+        """Sets data location.
+
+        :raises IOError: if directory does not exist.
+        """
+        data_loc = os.path.join(self.top_dirname, "data")
+        if os.path.isdir(data_loc) is False:
+            raise IOError("%s directory not found" % data_loc)
+        else:
+            self.data_loc = data_loc
+
+    def _build_network(self):
+        """Build network."""
+        reader = CSVReader(self.data_loc)
+        self.bus = reader.bus
+        self.plant = reader.plant
+        self.branch = reader.branch
+        self.dcline = reader.dcline
+        self.gencost["after"] = self.gencost["before"] = reader.gencost
+
+        self._add_information_to_model()
+
+        if self.umbrella_interconnect not in self.interconnect:
+            self._drop_interconnect()
+
+    def _add_information_to_model(self):
+        self.sub = csv_to_data_frame(self.data_loc, "sub.csv")
+        self.bus2sub = csv_to_data_frame(self.data_loc, "bus2sub.csv")
+        self.id2zone = csv_to_data_frame(self.data_loc, "zone.csv").zone_name.to_dict()
+        self.zone2id = {v: k for k, v in self.id2zone.items()}
+
+        add_zone_to_grid_data_frames(self)
+        add_coord_to_grid_data_frames(self)
+
+    def _drop_interconnect(self):
+        """Trim data frames to only keep information pertaining to the user
+        defined interconnect(s).
+
+        """
+        for key, value in self.__dict__.items():
+            if key in ["sub", "bus2sub", "bus", "plant", "branch"]:
+                value.query("interconnect == @self.interconnect", inplace=True)
+            elif key == "gencost":
+                value["before"].query(
+                    "interconnect == @self.interconnect", inplace=True
+                )
+            elif key == "dcline":
+                value.query(
+                    "from_interconnect == @self.interconnect &"
+                    "to_interconnect == @self.interconnect",
+                    inplace=True,
+                )
+        self.id2zone = {k: self.id2zone[k] for k in self.bus.zone_id.unique()}
+        self.zone2id = {value: key for key, value in self.id2zone.items()}
 
 
 def storage_template():

powersimdata/input/scenario_grid.py

@@ -21,21 +21,19 @@ def __init__(self, filename):
 
         :param str filename: path to file.
         """
+        self.filename = filename
         super().__init__()
-        self._set_data_loc(filename)
 
-        self._build_network()
-
-    def _set_data_loc(self, filename):
+    def _set_data_loc(self):
         """Sets data location.
 
         :param str filename: path to file
         :raises FileNotFoundError: if file does not exist.
         """
-        if os.path.isfile(filename) is False:
-            raise FileNotFoundError("%s file not found" % filename)
+        if os.path.isfile(self.filename) is False:
+            raise FileNotFoundError("%s file not found" % self.filename)
         else:
-            self.data_loc = filename
+            self.data_loc = self.filename
 
     def _read_network(self):
         data = loadmat(self.data_loc, squeeze_me=True, struct_as_record=False)

powersimdata/network/hifld/model.py

@@ -1,72 +1,16 @@
 import os
 
 from powersimdata.input.abstract_grid import AbstractGrid
-from powersimdata.input.helpers import (
-    add_coord_to_grid_data_frames,
-    add_zone_to_grid_data_frames,
-    csv_to_data_frame,
-)
-from powersimdata.network.csv_reader import CSVReader
-from powersimdata.network.usa_tamu.constants.storage import defaults
+from powersimdata.network.hifld.constants.storage import defaults
 
 
 class HIFLD(AbstractGrid):
     def __init__(self, interconnect):
         """Constructor."""
-        super().__init__()
-        self._set_data_loc()
-
+        self.top_dirname = os.path.dirname(__file__)
         self.interconnect = check_and_format_interconnect(interconnect)
-        self._build_network()
-
-    def _set_data_loc(self):
-        """Sets data location.
-
-        :raises IOError: if directory does not exist.
-        """
-        top_dirname = os.path.dirname(__file__)
-        data_loc = os.path.join(top_dirname, "data")
-        if os.path.isdir(data_loc) is False:
-            raise IOError("%s directory not found" % data_loc)
-        else:
-            self.data_loc = data_loc
-
-    def _build_network(self):
-        """Build network."""
-        reader = CSVReader(self.data_loc)
-        self.bus = reader.bus
-        self.plant = reader.plant
-        self.branch = reader.branch
-        self.dcline = reader.dcline
-        self.gencost["after"] = self.gencost["before"] = reader.gencost
-
-        self.storage.update(defaults)
-
-        add_information_to_model(self)
-
-        if "USA" not in self.interconnect:
-            self._drop_interconnect()
-
-    def _drop_interconnect(self):
-        """Trim data frames to only keep information pertaining to the user
-        defined interconnect(s).
-
-        """
-        for key, value in self.__dict__.items():
-            if key in ["sub", "bus2sub", "bus", "plant", "branch"]:
-                value.query("interconnect == @self.interconnect", inplace=True)
-            elif key == "gencost":
-                value["before"].query(
-                    "interconnect == @self.interconnect", inplace=True
-                )
-            elif key == "dcline":
-                value.query(
-                    "from_interconnect == @self.interconnect &"
-                    "to_interconnect == @self.interconnect",
-                    inplace=True,
-                )
-        self.id2zone = {k: self.id2zone[k] for k in self.bus.zone_id.unique()}
-        self.zone2id = {value: key for key, value in self.id2zone.items()}
+        self.umbrella_interconnect = "USA"
+        super().__init__()
 
 
 def check_and_format_interconnect(interconnect):
@@ -103,17 +47,3 @@ def interconnect_to_name(interconnect):
     :param iterable interconnect: interconnect name(s).
     """
     return "_".join(sorted(check_and_format_interconnect(interconnect)))
-
-
-def add_information_to_model(model):
-    """Adds information to TAMU model. This is done inplace.
-
-    :param powersimdata.input.TAMU model: TAMU instance.
-    """
-    model.sub = csv_to_data_frame(model.data_loc, "sub.csv")
-    model.bus2sub = csv_to_data_frame(model.data_loc, "bus2sub.csv")
-    model.id2zone = csv_to_data_frame(model.data_loc, "zone.csv").zone_name.to_dict()
-    model.zone2id = {v: k for k, v in model.id2zone.items()}
-
-    add_zone_to_grid_data_frames(model)
-    add_coord_to_grid_data_frames(model)

powersimdata/network/usa_tamu/model.py

@@ -1,12 +1,6 @@
 import os
 
 from powersimdata.input.abstract_grid import AbstractGrid
-from powersimdata.input.helpers import (
-    add_coord_to_grid_data_frames,
-    add_zone_to_grid_data_frames,
-    csv_to_data_frame,
-)
-from powersimdata.network.csv_reader import CSVReader
 from powersimdata.network.usa_tamu.constants.storage import defaults
 
 
@@ -18,61 +12,12 @@ class TAMU(AbstractGrid):
 
     def __init__(self, interconnect):
         """Constructor."""
-        super().__init__()
-        self._set_data_loc()
-
+        self.top_dirname = os.path.dirname(__file__)
         self.interconnect = check_and_format_interconnect(interconnect)
-        self._build_network()
-
-    def _set_data_loc(self):
-        """Sets data location.
-
-        :raises IOError: if directory does not exist.
-        """
-        top_dirname = os.path.dirname(__file__)
-        data_loc = os.path.join(top_dirname, "data")
-        if os.path.isdir(data_loc) is False:
-            raise IOError("%s directory not found" % data_loc)
-        else:
-            self.data_loc = data_loc
-
-    def _build_network(self):
-        """Build network."""
-        reader = CSVReader(self.data_loc)
-        self.bus = reader.bus
-        self.plant = reader.plant
-        self.branch = reader.branch
-        self.dcline = reader.dcline
-        self.gencost["after"] = self.gencost["before"] = reader.gencost
-
+        self.umbrella_interconnect = "USA"
+        super().__init__()
         self.storage.update(defaults)
 
-        add_information_to_model(self)
-
-        if "USA" not in self.interconnect:
-            self._drop_interconnect()
-
-    def _drop_interconnect(self):
-        """Trim data frames to only keep information pertaining to the user
-        defined interconnect(s).
-
-        """
-        for key, value in self.__dict__.items():
-            if key in ["sub", "bus2sub", "bus", "plant", "branch"]:
-                value.query("interconnect == @self.interconnect", inplace=True)
-            elif key == "gencost":
-                value["before"].query(
-                    "interconnect == @self.interconnect", inplace=True
-                )
-            elif key == "dcline":
-                value.query(
-                    "from_interconnect == @self.interconnect &"
-                    "to_interconnect == @self.interconnect",
-                    inplace=True,
-                )
-        self.id2zone = {k: self.id2zone[k] for k in self.bus.zone_id.unique()}
-        self.zone2id = {value: key for key, value in self.id2zone.items()}
-
 
 def check_and_format_interconnect(interconnect):
     """Checks interconnect.
@@ -108,17 +53,3 @@ def interconnect_to_name(interconnect):
     :param list interconnect: interconnect name(s).
     """
     return "_".join(sorted(check_and_format_interconnect(interconnect)))
-
-
-def add_information_to_model(model):
-    """Adds information to TAMU model. This is done inplace.
-
-    :param powersimdata.input.TAMU model: TAMU instance.
-    """
-    model.sub = csv_to_data_frame(model.data_loc, "sub.csv")
-    model.bus2sub = csv_to_data_frame(model.data_loc, "bus2sub.csv")
-    model.id2zone = csv_to_data_frame(model.data_loc, "zone.csv").zone_name.to_dict()
-    model.zone2id = {v: k for k, v in model.id2zone.items()}
-
-    add_zone_to_grid_data_frames(model)
-    add_coord_to_grid_data_frames(model)