Infra x/dewa

disco/cli/pv_deployments.py

@@ -25,15 +25,15 @@
 logger = logging.getLogger(__name__)
 
 
-def create_pv_deployments(input_path: str, hierarchy: str, config: dict):
+def create_pv_deployments(input_path: str, hierarchy: str, config: dict, hv_min: float, hv_max: float, **kwargs):
     """A method for generating pv deployments"""
     hierarchy = DeploymentHierarchy(hierarchy)
     config = SimpleNamespace(**config)
     if not config.placement:
         print(f"'-p' or '--placement' should not be None for this action, choose from {PLACEMENT_CHOICE}")
         sys.exit()
     manager = PVDeploymentManager(input_path, hierarchy, config)
-    summary = manager.generate_pv_deployments()
+    summary = manager.generate_pv_deployments(hv_min=hv_min, hv_max=hv_max, **kwargs)
     print(json.dumps(summary, indent=2))
 
 
@@ -289,6 +289,31 @@ def pv_deployments():
     default=random.randint(1, 1000000),
     help="Set an initial integer seed for making PV deployments reproducible"
 )
+@click.option(
+    "-min-hv", "--minimum-high-voltage",
+    type=click.FLOAT,
+    default=1,
+    show_default=True,
+    help="Minimum voltage level for high voltage buses.",
+)
+@click.option(
+    "-max-hv", "--maximum-high-voltage",
+    type=click.FLOAT,
+    default=None,
+    help="Maximum voltage level for high voltage buses.",
+)
+@click.option(
+    "-large-pv-max", "--large-pv-upper-bound",
+    type=click.FLOAT,
+    default=None,
+    help="Upper bound for large PV power.",
+)
+@click.option(
+    "-small-pv-max", "--small-pv-upper-bound",
+    type=click.FLOAT,
+    default=None,
+    help="Upper bound for small PV power.",
+)
 @click.option(
     "--verbose",
     type=click.BOOL,
@@ -316,6 +341,10 @@ def source_tree_1(
     pv_upscale,
     pv_deployments_dirname,
     random_seed,
+    hv_min,
+    hv_max,
+    large_pv_upper_bound,
+    small_pv_upper_bound,
     verbose
 ):
     """Generate PV deployments for source tree 1."""
@@ -345,10 +374,21 @@ def source_tree_1(
     }
     action_function = ACTION_MAPPING[action]
     args = [input_path, hierarchy, config]
+    kwargs = {}
     if action == "create-configs":
         args.append(control_name)
         args.append(kw_limit)
-    action_function(*args)
+    if action == "create-pv":
+        args.append(hv_min)
+        args.append(hv_max)
+
+    if large_pv_upper_bound:
+        kwargs['large_pv_upper_bound'] = large_pv_upper_bound
+
+    if small_pv_upper_bound:
+        kwargs['small_pv_upper_bound'] = small_pv_upper_bound
+
+    action_function(*args, **kwargs)
 
 
 pv_deployments.add_command(source_tree_1)

disco/sources/source_tree_1/pv_deployments.py

@@ -237,7 +237,7 @@ def get_customer_distance(self) -> SimpleNamespace:
             flag = dss.Loads.Next()
         return result
 
-    def get_highv_buses(self, kv_min: int = 1) -> SimpleNamespace:
+    def get_highv_buses(self, kv_min: float = 1, kv_max: float = None) -> SimpleNamespace:
         """Return highv buses"""
         result = SimpleNamespace(bus_kv={}, hv_bus_distance={})
         flag = dss.Lines.First()
@@ -246,7 +246,7 @@ def get_highv_buses(self, kv_min: int = 1) -> SimpleNamespace:
             for bus in buses:
                 dss.Circuit.SetActiveBus(bus)
                 kvbase = dss.Bus.kVBase()
-                if kvbase >= kv_min:
+                if kvbase >= kv_min and (kv_max is None or kvbase <= kv_max):
                     result.bus_kv[bus] = dss.Bus.kVBase()
                     result.hv_bus_distance[bus] = dss.Bus.Distance()
             flag = dss.Lines.Next()
@@ -356,7 +356,7 @@ def load_pvdss_instance(self) -> PVDSSInstance:
             raise
         return pvdss_instance
 
-    def deploy_all_pv_scenarios(self) -> dict:
+    def deploy_all_pv_scenarios(self, hv_min, hv_max, **kwargs) -> dict:
         """Given a feeder path, generate all PV scenarios for the feeder"""
         feeder_name = self.get_feeder_name()
         pvdss_instance = self.load_pvdss_instance()
@@ -372,7 +372,14 @@ def deploy_all_pv_scenarios(self) -> dict:
 
         # combined bus distance
         customer_distance = pvdss_instance.get_customer_distance()
-        highv_buses = pvdss_instance.get_highv_buses()
+        highv_buses = pvdss_instance.get_highv_buses(kv_min=hv_min, kv_max=hv_max)
+
+        # Filter out overlapping buses from customer_distance
+        customer_distance.bus_distance = {
+            bus: dist for bus, dist in customer_distance.bus_distance.items()
+            if bus not in highv_buses.hv_bus_distance
+        }
+
         combined_bus_distance = pvdss_instance.combine_bus_distances(customer_distance, highv_buses)
         if max(combined_bus_distance.values()) == 0:
             logger.warning(
@@ -420,9 +427,9 @@ def deploy_all_pv_scenarios(self) -> dict:
                     bus_kv=highv_buses.bus_kv,
                     pv_records=pv_records,
                     penetration=penetration,
-                    sample=sample
+                    sample=sample,
                 )
-                existing_pv, pv_records = self.deploy_pv_scenario(data)
+                existing_pv, pv_records = self.deploy_pv_scenario(data, **kwargs)
 
         return feeder_stats.__dict__
 
@@ -456,7 +463,7 @@ def get_pv_systems_file(self, sample: int, penetration: int) -> str:
         pv_systems_file = os.path.join(penetration_path, PV_SYSTEMS_FILENAME)
         return pv_systems_file
 
-    def deploy_pv_scenario(self, data: SimpleNamespace) -> dict:
+    def deploy_pv_scenario(self, data: SimpleNamespace, **kwargs) -> dict:
         """Generate PV deployments dss file in scenario
 
         Parameters
@@ -496,7 +503,7 @@ def deploy_pv_scenario(self, data: SimpleNamespace) -> dict:
                             if base_min_pv_size > 0:
                                 continue
                             min_pv_size = existing_pv[bus]
-                            max_pv_size = self.get_maximum_pv_size(bus, data)
+                            max_pv_size = self.get_maximum_pv_size(bus, data, **kwargs)
                             random_pv_size = self.generate_pv_size_from_pdf(min_pv_size, max_pv_size)
                             pv_size = min(random_pv_size, min_pv_size + remaining_pv_to_install)
                             pv_added_capacity = pv_size - min_pv_size
@@ -546,7 +553,7 @@ def deploy_pv_scenario(self, data: SimpleNamespace) -> dict:
                     if (base_min_pv_size > 0 or min_pv_size > 0) and (not self.config.pv_upscale):
                         pass
                     else:
-                        max_pv_size = self.get_maximum_pv_size(picked_candidate, data)
+                        max_pv_size = self.get_maximum_pv_size(picked_candidate, data, **kwargs)
                         random_pv_size = self.generate_pv_size_from_pdf(0, max_pv_size)
                         pv_size = min(random_pv_size, remaining_pv_to_install)
                         pv_string = self.add_pv_string(picked_candidate, pv_type.value, pv_size, pv_string)
@@ -696,6 +703,10 @@ def write_pv_string(self, pv_string: str, data: SimpleNamespace) -> None:
 
     def get_pv_bus_subset(self, bus_distance: dict, subset_idx: int, priority_buses: list) -> list:
         """Return candidate buses"""
+        if not bus_distance:
+            logger.warning("bus_distance is empty. Returning an empty candidate_bus_array.")
+            return []
+
         max_dist = max(bus_distance.values())
         min_dist = min(bus_distance.values())
         if self.config.placement == Placement.CLOSE.value:
@@ -977,7 +988,8 @@ def get_categorical_remaining_pvs(self, data: SimpleNamespace) -> dict:
 
     @classmethod
     def get_maximum_pv_size(cls, bus: str, data: SimpleNamespace, **kwargs) -> int:
-        max_bus_pv_size = 100 * random.randint(1, 50)
+        upper_bound = kwargs.get('large_pv_upper_bound', 50)
+        max_bus_pv_size = 100 * random.randint(1, upper_bound)
         return max_bus_pv_size
 
 
@@ -1002,14 +1014,17 @@ def get_maximum_pv_size(cls, bus: str, data: SimpleNamespace, max_load_factor: f
         customer_annual_kwh = kwargs.get("customer_annual_kwh", {})
         annual_sun_hours = kwargs.get("annual_sun_hours", None)
 
-        pv_size_array = [max_load_factor * data.bus_totalload[bus]]
-        if roof_area and pv_efficiency:
-            value = roof_area[bus] * pv_efficiency
-            pv_size_array.append(value)
-        if customer_annual_kwh and annual_sun_hours:
-            value = customer_annual_kwh[bus] / annual_sun_hours
-            pv_size_array.append(value)
-        max_bus_pv_size = min(pv_size_array)
+        if 'small_pv_upper_bound' in kwargs:
+            max_bus_pv_size = [kwargs['small_pv_upper_bound']]
+        else:
+            pv_size_array = [max_load_factor * data.bus_totalload[bus]]
+            if roof_area and pv_efficiency:
+                value = roof_area[bus] * pv_efficiency
+                pv_size_array.append(value)
+            if customer_annual_kwh and annual_sun_hours:
+                value = customer_annual_kwh[bus] / annual_sun_hours
+                pv_size_array.append(value)
+            max_bus_pv_size = min(pv_size_array)
         return max_bus_pv_size
 
 
@@ -1584,7 +1599,7 @@ def __init__(self, input_path: str, hierarchy: DeploymentHierarchy, config: Simp
         """
         super().__init__(input_path, hierarchy, config)
 
-    def generate_pv_deployments(self) -> dict:
+    def generate_pv_deployments(self, hv_min: float = 1, hv_max: float = None, **kwargs) -> dict:
         """Given input path, generate pv deployments"""
         summary = {}
         feeder_paths = self.get_feeder_paths()
@@ -1594,7 +1609,7 @@ def generate_pv_deployments(self) -> dict:
                 "Set initial integer seed %s for PV deployments on feeder - %s",
                 self.config.random_seed, feeder_path
             )
-            feeder_stats = generator.deploy_all_pv_scenarios()
+            feeder_stats = generator.deploy_all_pv_scenarios(hv_min, hv_max, **kwargs)
             summary[feeder_path] = feeder_stats
         return summary