Merge branch 'project/411_LoMa_14aOptimization_with_virtual_generators' into features/#618-create-interface-to-import-all-LoMa-data-for-14a-optimization

Author: CarlosEpia

edisgo/edisgo.py

@@ -35,9 +35,12 @@
 from edisgo.io.ding0_import import import_ding0_grid
 from edisgo.io.electromobility_import import (
     distribute_charging_demand,
+    distribute_charging_demand_14a, #engine
     import_electromobility_from_dir,
     import_electromobility_from_oedb,
+    import_electromobility_from_oedb_14a,
     integrate_charging_parks,
+    integrate_charging_parks_14a,
 )
 from edisgo.io.heat_pump_import import oedb as import_heat_pumps_oedb
 from edisgo.io.storage_import import home_batteries_oedb
@@ -53,7 +56,7 @@
 from edisgo.opf.results.opf_result_class import OPFResults
 from edisgo.tools import plots, tools
 from edisgo.tools.config import Config
-from edisgo.tools.geo import find_nearest_bus
+from edisgo.tools.geo import find_nearest_bus, find_nearest_bus_14a
 from edisgo.tools.spatial_complexity_reduction import spatial_complexity_reduction
 from edisgo.tools.tools import determine_grid_integration_voltage_level
 
@@ -1804,6 +1807,91 @@ def integrate_component_based_on_geolocation(
                 )
 
         return comp_name
+    
+    def integrate_component_based_on_geolocation_14a(
+        self,
+        comp_type,
+        geolocation,
+        voltage_level=None,
+        add_ts=True,
+        ts_active_power=None,
+        ts_reactive_power=None,
+        **kwargs,
+    ):
+        """
+        14a variant of integrate_component_based_on_geolocation.
+        
+        Uses explicit _14a helper functions where custom behavior was introduced,
+        while leaving the original integrate_component_based_on_geolocation
+        unchanged for standard users.
+        """
+        supported_voltage_levels = {4, 5, 6, 7}
+        p_nom = kwargs.get("p_nom", None)
+        p_set = kwargs.get("p_set", None)
+        
+        p = p_nom if p_set is None else p_set
+        kwargs["p"] = p
+
+        if voltage_level not in supported_voltage_levels:
+            if p is None:
+                raise ValueError(
+                    "Neither appropriate voltage level nor nominal power were supplied."
+                )
+            voltage_level = determine_grid_integration_voltage_level(self, p)
+
+        # convert geolocation to shapely Point if needed
+        if type(geolocation) is not Point:
+            geolocation = Point(geolocation)
+
+        kwargs["geom"] = geolocation
+        kwargs["voltage_level"] = voltage_level
+
+        # -------------------------
+        # Connect in MV
+        # -------------------------
+        if voltage_level in [4, 5]:
+            comp_name = self.topology.connect_to_mv_14a(self, kwargs, comp_type)
+
+        # -------------------------
+        # Connect in LV
+        # -------------------------
+        else:
+            lv_buses = self.topology.buses_df.drop(self.topology.mv_grid.buses_df.index)
+            lv_buses_dropna = lv_buses.dropna(axis=0, subset=["x", "y"])
+
+            # fallback path for non-georeferenced LV grids
+            if len(lv_buses_dropna) < len(lv_buses):
+                if kwargs.get("mvlv_subst_id", None) is None:
+                    substations = self.topology.buses_df.loc[
+                        self.topology.transformers_df.bus1.unique()
+                    ]
+                    nearest_substation, _ = find_nearest_bus_14a(geolocation, substations)
+                    kwargs["mvlv_subst_id"] = int(nearest_substation.split("_")[-2])
+
+                comp_name = self.topology.connect_to_lv_14a(self, kwargs, comp_type)
+
+            else:
+                max_distance_from_target_bus = kwargs.pop(
+                    "max_distance_from_target_bus", 0.3 #CHANGED #14a
+                )
+                comp_name = self.topology.connect_to_lv_based_on_geolocation_14a(
+                    self, kwargs, comp_type, max_distance_from_target_bus
+                )
+
+        if add_ts:
+            if comp_type == "generator":
+                self.set_time_series_manual(
+                    generators_p=pd.DataFrame({comp_name: ts_active_power}),
+                    generators_q=pd.DataFrame({comp_name: ts_reactive_power}),
+                )
+            else:
+                self.set_time_series_manual(
+                    loads_p=pd.DataFrame({comp_name: ts_active_power}),
+                    loads_q=pd.DataFrame({comp_name: ts_reactive_power}),
+                )
+
+        return comp_name
+
 
     def remove_component(self, comp_type, comp_name, drop_ts=True):
         """
@@ -2111,6 +2199,58 @@ def import_electromobility(
 
         integrate_charging_parks(self)
 
+    def import_electromobility_14a(
+        self,
+        data_source: str = "oedb",
+        scenario: str = None,
+        import_electromobility_data_kwds=None,
+        allocate_charging_demand_kwds=None,
+    ):
+        """
+        14a variant of import_electromobility.
+        
+        This method uses the specific 14a EV integration path.
+        """
+        if data_source != "oedb":
+            raise ValueError(
+                "Invalid input for parameter 'data_source'. Currently only 'oedb' is supported."
+            )
+
+        valid_scenarios = {"eGon2035", "eGon100RE"}
+        if scenario not in valid_scenarios:
+            raise ValueError(
+                f"Invalid scenario '{scenario}'. Possible options are {sorted(valid_scenarios)}."
+            )
+            
+        if self.engine is None:
+            raise ValueError(
+                "No database engine available. Please set 'self.engine' before calling "
+                "import_electromobility_14a()."
+            )
+
+        if import_electromobility_data_kwds is None:
+            import_electromobility_data_kwds = {}
+
+        if "shapefile_path" not in import_electromobility_data_kwds:
+            raise ValueError(
+                "For import_electromobility_14a with data_source='oedb', "
+                "'shapefile_path' must be provided in import_electromobility_data_kwds."
+            )
+
+        if allocate_charging_demand_kwds is None:
+            allocate_charging_demand_kwds = {}
+        
+        import_electromobility_from_oedb_14a(
+            self,
+            scenario=scenario,
+            engine=self.engine,
+            **import_electromobility_data_kwds,
+        )
+
+        distribute_charging_demand_14a(self, **allocate_charging_demand_kwds)
+        
+        integrate_charging_parks_14a(self)
+
     def apply_charging_strategy(self, strategy="dumb", **kwargs):
         """
         Applies charging strategy to set EV charging time series at charging parks.