Implement mview replacement functions and filters

edisgo/io/generators_import.py

@@ -25,6 +25,7 @@
     import geopandas as gpd
 
     from egoio.db_tables import model_draft, supply
+    from edisgo.io.mviews_filters import build_conv_scenario_filter, build_res_scenario_filter
     from shapely.ops import transform
     from shapely.wkt import loads as wkt_loads
 
@@ -117,22 +118,22 @@ def _import_conv_generators(session):
         # build query
         generators_sqla = (
             session.query(
-                orm_conv_generators.columns.id,
-                orm_conv_generators.columns.id.label("generator_id"),
-                orm_conv_generators.columns.subst_id,
-                orm_conv_generators.columns.la_id,
-                orm_conv_generators.columns.capacity.label("p_nom"),
-                orm_conv_generators.columns.voltage_level,
-                orm_conv_generators.columns.fuel.label("generator_type"),
+                orm_conv_generators.id,
+                orm_conv_generators.id.label("generator_id"),
+                orm_conv_generators.subst_id,
+                orm_conv_generators.la_id,
+                orm_conv_generators.capacity.label("p_nom"),
+                orm_conv_generators.voltage_level,
+                orm_conv_generators.fuel.label("generator_type"),
                 func.ST_AsText(
-                    func.ST_Transform(orm_conv_generators.columns.geom, srid)
+                    func.ST_Transform(orm_conv_generators.geom, srid)
                 ).label("geom"),
             )
             .filter(
-                orm_conv_generators.columns.subst_id
+                orm_conv_generators.subst_id
                 == edisgo_object.topology.mv_grid.id
             )
-            .filter(orm_conv_generators.columns.voltage_level.in_([4, 5]))
+            .filter(orm_conv_generators.voltage_level.in_([4, 5]))
             .filter(orm_conv_generators_version)
         )
 
@@ -161,32 +162,32 @@ def _import_res_generators(session):
         # build basic query
         generators_sqla = (
             session.query(
-                orm_re_generators.columns.id,
-                orm_re_generators.columns.id.label("generator_id"),
-                orm_re_generators.columns.subst_id,
-                orm_re_generators.columns.la_id,
-                orm_re_generators.columns.mvlv_subst_id,
-                orm_re_generators.columns.electrical_capacity.label("p_nom"),
-                orm_re_generators.columns.generation_type.label("generator_type"),
-                orm_re_generators.columns.generation_subtype.label("subtype"),
-                orm_re_generators.columns.voltage_level,
-                orm_re_generators.columns.w_id.label("weather_cell_id"),
+                orm_re_generators.id,
+                orm_re_generators.id.label("generator_id"),
+                orm_re_generators.subst_id,
+                orm_re_generators.la_id,
+                orm_re_generators.mvlv_subst_id,
+                orm_re_generators.electrical_capacity.label("p_nom"),
+                orm_re_generators.generation_type.label("generator_type"),
+                orm_re_generators.generation_subtype.label("subtype"),
+                orm_re_generators.voltage_level,
+                orm_re_generators.w_id.label("weather_cell_id"),
                 func.ST_AsText(
-                    func.ST_Transform(orm_re_generators.columns.rea_geom_new, srid)
+                    func.ST_Transform(orm_re_generators.rea_geom_new, srid)
                 ).label("geom"),
                 func.ST_AsText(
-                    func.ST_Transform(orm_re_generators.columns.geom, srid)
+                    func.ST_Transform(orm_re_generators.geom, srid)
                 ).label("geom_em"),
             )
             .filter(
-                orm_re_generators.columns.subst_id == edisgo_object.topology.mv_grid.id
+                orm_re_generators.subst_id == edisgo_object.topology.mv_grid.id
             )
             .filter(orm_re_generators_version)
         )
 
         # extend basic query for MV generators and read data from db
         generators_mv_sqla = generators_sqla.filter(
-            orm_re_generators.columns.voltage_level.in_([4, 5])
+            orm_re_generators.voltage_level.in_([4, 5])
         )
         gens_mv = pd.read_sql_query(
             generators_mv_sqla.statement, session.bind, index_col="id"
@@ -200,7 +201,7 @@ def _import_res_generators(session):
 
         # extend basic query for LV generators and read data from db
         generators_lv_sqla = generators_sqla.filter(
-            orm_re_generators.columns.voltage_level.in_([6, 7])
+            orm_re_generators.voltage_level.in_([6, 7])
         )
         gens_lv = pd.read_sql_query(
             generators_lv_sqla.statement, session.bind, index_col="id"
@@ -302,39 +303,45 @@ def _validate_sample_geno_location():
     oedb_data_source = edisgo_object.config["data_source"]["oedb_data_source"]
     srid = edisgo_object.topology.grid_district["srid"]
 
-    # load ORM names
-    orm_conv_generators_name = (
-        edisgo_object.config[oedb_data_source]["conv_generators_prefix"]
-        + generator_scenario
-        + edisgo_object.config[oedb_data_source]["conv_generators_suffix"]
-    )
-    orm_re_generators_name = (
-        edisgo_object.config[oedb_data_source]["re_generators_prefix"]
-        + generator_scenario
-        + edisgo_object.config[oedb_data_source]["re_generators_suffix"]
+    # Map generator_scenario to full scenario names for mview filter logic
+    scenario_mapping = {
+        'nep2035': 'NEP 2035',
+        'ego100': 'eGo 100',
+        'status_quo': 'Status Quo',
+        'sq': 'Status Quo'
+    }
+    scenario_name = scenario_mapping.get(
+        generator_scenario.lower(),
+        generator_scenario
     )
 
     if oedb_data_source == "model_draft":
-        # import ORMs
-        orm_conv_generators = model_draft.__getattribute__(orm_conv_generators_name)
-        orm_re_generators = model_draft.__getattribute__(orm_re_generators_name)
+        # Use base tables from model_draft schema (CamelCase ORM names, without _mview suffix)
+        orm_conv_generators = model_draft.__getattribute__("EgoDpSupplyConvPowerplant")
+        orm_re_generators = model_draft.__getattribute__("EgoDpSupplyResPowerplant")
 
-        # set dummy version condition (select all generators)
-        orm_conv_generators_version = 1 == 1
-        orm_re_generators_version = 1 == 1
+        # Build scenario filters that replicate mview logic
+        orm_conv_generators_version = build_conv_scenario_filter(
+            orm_conv_generators, scenario_name, version=None
+        )
+        orm_re_generators_version = build_res_scenario_filter(
+            orm_re_generators, scenario_name, version=None
+        )
 
     elif oedb_data_source == "versioned":
         data_version = edisgo_object.config["versioned"]["version"]
 
-        # import ORMs
-        orm_conv_generators = supply.__getattribute__(orm_conv_generators_name)
-        orm_re_generators = supply.__getattribute__(orm_re_generators_name)
+        # Use base tables from supply schema (CamelCase ORM names, without _mview suffix)
+        orm_conv_generators = supply.__getattribute__("EgoDpConvPowerplant")
+        orm_re_generators = supply.__getattribute__("EgoDpResPowerplant")
 
-        # set version condition
-        orm_conv_generators_version = (
-            orm_conv_generators.columns.version == data_version
+        # Build scenario filters that replicate mview logic with version
+        orm_conv_generators_version = build_conv_scenario_filter(
+            orm_conv_generators, scenario_name, version=data_version
+        )
+        orm_re_generators_version = build_res_scenario_filter(
+            orm_re_generators, scenario_name, version=data_version
         )
-        orm_re_generators_version = orm_re_generators.columns.version == data_version
 
     # get conventional and renewable generators
     with session_scope() as session:

edisgo/io/mviews_filters.py

@@ -0,0 +1,159 @@
+"""
+Helper functions to build scenario filters that replicate mview logic.
+
+These functions create SQLAlchemy filter conditions that match the original
+materialized view definitions, allowing direct queries on base tables.
+
+Author: Generated for mviews replacement
+Date: 2025-12-17
+"""
+
+
+def build_conv_scenario_filter(orm_table, scenario, version=None, preversion='v0.3.0'):
+    """
+    Build filter conditions for conventional power plants matching mview logic.
+
+    Parameters
+    ----------
+    orm_table : SQLAlchemy Table
+        The ORM table object (e.g., ego_dp_conv_powerplant)
+    scenario : str
+        Scenario name ('NEP 2035', 'eGo 100', 'Status Quo')
+    version : str or list, optional
+        Version filter. If None, uses defaults based on scenario.
+    preversion : str, default 'v0.3.0'
+        Preversion filter
+
+    Returns
+    -------
+    SQLAlchemy filter condition
+        Combined filter matching the mview definition
+    """
+    # Set default versions
+    if version is None:
+        if scenario in ['NEP 2035', 'eGo 100']:
+            versions = ['v0.4.2', 'v0.4.4', 'v0.4.5']
+        else:
+            versions = ['v0.4.5']
+    else:
+        versions = [version] if isinstance(version, str) else version
+
+    # Base filters
+    filters = [
+        orm_table.capacity > 0,
+        orm_table.preversion == preversion,
+    ]
+
+    # Version filter
+    if hasattr(orm_table, 'version'):
+        filters.append(orm_table.version.in_(versions))
+
+    # Scenario-specific filters
+    if scenario == 'eGo 100':
+        # eGo 100: only pumped storage from NEP 2035
+        filters.extend([
+            orm_table.scenario == 'NEP 2035',
+            orm_table.fuel == 'pumped_storage',
+            (orm_table.shutdown == None) |
+            (orm_table.shutdown >= 2049)
+        ])
+    elif scenario == 'NEP 2035':
+        # NEP 2035: exclude hydro, filter by shutdown
+        # Note: OEP doesn't support NOT IN, so we use individual != filters
+        filters.extend([
+            orm_table.scenario == 'NEP 2035',
+            orm_table.fuel != 'hydro',
+            orm_table.fuel != 'run_of_river',
+            orm_table.fuel != 'reservoir',
+            (orm_table.shutdown == None) |
+            (orm_table.shutdown >= 2034)
+        ])
+    else:
+        # Generic scenario filter
+        filters.append(orm_table.scenario == scenario)
+
+    # Combine all filters with AND
+    from sqlalchemy import and_
+    return and_(*filters)
+
+
+def build_res_scenario_filter(orm_table, scenario, version=None, preversion='v0.3.0'):
+    """
+    Build filter conditions for renewable power plants matching mview logic.
+
+    IMPORTANT: NEP 2035 mview is a UNION of Status Quo + NEP 2035 generators!
+    The mview definition shows that ego_dp_res_powerplant_nep2035_mview contains
+    BOTH Status Quo (solar/wind only, excl. offshore) AND NEP 2035 generators.
+
+    Parameters
+    ----------
+    orm_table : SQLAlchemy Table
+        The ORM table object (e.g., ego_dp_res_powerplant)
+    scenario : str
+        Scenario name ('NEP 2035', 'eGo 100', 'Status Quo')
+    version : str or list, optional
+        Version filter. If None, uses defaults based on scenario.
+    preversion : str, default 'v0.3.0'
+        Preversion filter
+
+    Returns
+    -------
+    SQLAlchemy filter condition
+        Combined filter matching the mview definition
+    """
+    from sqlalchemy import and_, or_
+
+    # Set default versions
+    if version is None:
+        if scenario == 'Status Quo':
+            versions = ['v0.4.4', 'v0.4.5']
+        else:
+            # NEP 2035 uses v0.4.4 and v0.4.5
+            versions = ['v0.4.4', 'v0.4.5']
+    else:
+        versions = [version] if isinstance(version, str) else version
+
+    # Base filters (always required)
+    base_filters = [
+        orm_table.electrical_capacity > 0,
+        orm_table.preversion == preversion,
+    ]
+
+    # Version filter
+    if hasattr(orm_table, 'version'):
+        base_filters.append(orm_table.version.in_(versions))
+
+    # Scenario-specific filters
+    if scenario == 'NEP 2035':
+        # NEP 2035 mview is UNION of:
+        # 1. Status Quo generators (solar/wind only, no offshore)
+        # 2. NEP 2035 generators (all types)
+        status_quo_filters = [
+            orm_table.scenario == 'Status Quo',
+            orm_table.generation_type.in_(['solar', 'wind']),
+            orm_table.generation_subtype != 'wind_offshore'
+        ]
+        nep2035_filters = [
+            orm_table.scenario == 'NEP 2035'
+        ]
+
+        # Combine: (Status Quo conditions) OR (NEP 2035 conditions)
+        scenario_filter = or_(
+            and_(*status_quo_filters),
+            and_(*nep2035_filters)
+        )
+        base_filters.append(scenario_filter)
+
+    elif scenario == 'Status Quo':
+        # Status Quo: only solar and wind (excluding offshore)
+        base_filters.extend([
+            orm_table.scenario == 'Status Quo',
+            orm_table.generation_type.in_(['solar', 'wind']),
+            orm_table.generation_subtype != 'wind_offshore'
+        ])
+    else:
+        # Generic scenario filter for other scenarios (like eGo 100)
+        base_filters.append(orm_table.scenario == scenario)
+
+    # Combine all filters with AND
+    return and_(*base_filters)