feature/validation_integration

pyproject.toml

@@ -47,6 +47,7 @@ dependencies = [
   "cdsapi",
   "click<8.1",
   "disaggregator @ git+https://github.com/openego/disaggregator.git@features/update-cache-directory#egg=disaggregator",
+  "egon-validation @ git+https://github.com/sagemaso/eGon-validation.git@v1.2.1",
   "entsoe-py>=0.6.2",
   "fiona==1.9.6",
   "Flask-Session<0.6.0",

src/egon/data/airflow/dags/pipeline.py

@@ -102,6 +102,9 @@
 from egon.data.datasets.zensus_vg250 import ZensusVg250
 from egon.data.metadata import Json_Metadata
 
+from egon.data.datasets.validation_report import ValidationReport
+from egon.data.datasets.final_validations import FinalValidations
+
 # Set number of threads used by numpy and pandas
 set_numexpr_threads()
 
@@ -730,6 +733,32 @@
             ]
         )
 
+    with TaskGroup(group_id="final_validations") as final_validations_group:
+        # Cross-cutting validations that check data consistency across datasets
+        # These run after all data generation but before the validation report
+        final_validations = FinalValidations(
+            dependencies=[
+                insert_data_ch4_storages,  # CH4Storages - for CH4 store validation
+                insert_H2_storage,          # HydrogenStoreEtrago - for H2 saltcavern validation
+                storage_etrago,             # StorageEtrago - general storage validation
+                hts_etrago_table,
+                fill_etrago_generators,
+                household_electricity_demand_annual,
+                cts_demand_buildings,
+                emobility_mit,
+                low_flex_scenario,
+            ]
+        )
+
+    with TaskGroup(group_id="validation_report") as validation_report_group:
+        # Generate validation report from all validation tasks
+        # Runs after all validations (including final_validations) are complete
+        validation_report = ValidationReport(
+            dependencies=[
+                final_validations,           # Wait for final validations
+            ]
+        )
+
     with TaskGroup(group_id="sanity_checks") as sanity_checks_group:
         # ########## Keep this dataset at the end
         # Sanity Checks

src/egon/data/datasets/DSM_cts_ind.py

@@ -32,6 +32,10 @@
     sources,
 )
 
+from egon_validation import (
+    ArrayCardinalityValidation
+)
+
 # CONSTANTS
 # TODO: move to datasets.yml
 CON = db.engine()
@@ -142,6 +146,35 @@ def __init__(self, dependencies):
             version=self.version,
             dependencies=dependencies,
             tasks=(dsm_cts_ind_processing,),
+            validation={
+                "data-quality": [
+                    ArrayCardinalityValidation(
+                        table="demand.egon_demandregio_sites_ind_electricity_dsm_timeseries",
+                        rule_id="ARRAY_VALIDATION.egon_demandregio_sites_ind_electricity_dsm_timeseries",
+                        array_column="p_set",
+                        expected_length=8760,
+                    ),
+                    ArrayCardinalityValidation(
+                        table="demand.egon_etrago_electricity_cts_dsm_timeseries",
+                        rule_id="ARRAY_VALIDATION.egon_etrago_electricity_cts_dsm_timeseries",
+                        array_column="p_set",
+                        expected_length=8760,
+                    ),
+                    ArrayCardinalityValidation(
+                        table="demand.egon_osm_ind_load_curves_individual_dsm_timeseries",
+                        rule_id="ARRAY_VALIDATION.egon_osm_ind_load_curves_individual_dsm_timeseries",
+                        array_column="p_set",
+                        expected_length=8760,
+                    ),
+                    ArrayCardinalityValidation(
+                        table="demand.egon_sites_ind_load_curves_individual_dsm_timeseries",
+                        rule_id="ARRAY_VALIDATION.egon_sites_ind_load_curves_individual_dsm_timeseries",
+                        array_column="p_set",
+                        expected_length=8760,
+                    ),
+                ]
+            },
+            on_validation_failure="continue"
         )
 
 

src/egon/data/datasets/__init__.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections import abc
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from functools import partial, reduce, update_wrapper
 from typing import Callable, Iterable, Set, Tuple, Union
 import re
@@ -12,9 +12,17 @@
 from airflow.operators.python import PythonOperator
 from sqlalchemy import Column, ForeignKey, Integer, String, Table, orm, tuple_
 from sqlalchemy.ext.declarative import declarative_base
+from typing import Dict, List
+from egon.data.validation import create_validation_tasks
 
 from egon.data import config, db, logger
 
+try:
+    from egon_validation.rules.base import Rule
+except ImportError:
+    Rule = None  # Type hint only
+
+
 Base = declarative_base()
 SCHEMA = "metadata"
 
@@ -197,6 +205,8 @@ class Dataset:
     #: The tasks of this :class:`Dataset`. A :class:`TaskGraph` will
     #: automatically be converted to :class:`Tasks_`.
     tasks: Tasks = ()
+    validation: Dict[str, List] = field(default_factory=dict)
+    on_validation_failure: str = "continue"
 
     def check_version(self, after_execution=()):
         scenario_names = config.settings()["egon-data"]["--scenarios"]
@@ -264,6 +274,27 @@ def __post_init__(self):
         self.dependencies = list(self.dependencies)
         if not isinstance(self.tasks, Tasks_):
             self.tasks = Tasks_(self.tasks)
+            # Process validation configuration
+        if self.validation:
+            validation_tasks = create_validation_tasks(
+                validation_dict=self.validation,
+                dataset_name=self.name,
+                on_failure=self.on_validation_failure
+            )
+
+            # Append validation tasks to existing tasks
+            if validation_tasks:
+                if hasattr(self.tasks, 'graph'):
+                    graph = self.tasks.graph
+                else:
+                    graph = self.tasks
+                if isinstance(graph, (tuple, set, list)):
+                    task_list = list(graph)
+                else:
+                    task_list = [graph]
+                task_list.extend(validation_tasks)
+                self.tasks = Tasks_(tuple(task_list))
+
         if len(self.tasks.last) > 1:
             # Explicitly create single final task, because we can't know
             # which of the multiple tasks finishes last.
@@ -302,3 +333,33 @@ def __post_init__(self):
         for p in predecessors:
             for first in self.tasks.first:
                 p.set_downstream(first)
+
+        # Link validation tasks to run after data tasks
+        if self.validation and validation_tasks:
+            # Get last non-validation tasks
+            non_validation_task_ids = [
+                task.task_id for task in self.tasks.values()
+                if not any(
+                    task.task_id.endswith(f".validate.{name}")
+                    for name in self.validation.keys()
+                )
+            ]
+
+            last_data_tasks = [
+                task for task in self.tasks.values()
+                if task.task_id in non_validation_task_ids
+                and task in self.tasks.last
+            ]
+
+            if not last_data_tasks:
+                # Fallback to last non-validation task
+                last_data_tasks = [
+                    task for task in self.tasks.values()
+                    if task.task_id in non_validation_task_ids
+                ][-1:]
+
+            # Link each validation task downstream of last data tasks
+            for validation_task in validation_tasks:
+                for last_task in last_data_tasks:
+                    last_task.set_downstream(validation_task)
+

src/egon/data/datasets/chp/__init__.py

@@ -47,6 +47,15 @@
     sources,
 )
 
+from egon_validation import (
+    RowCountValidation,
+    DataTypeValidation,
+    NotNullAndNotNaNValidation,
+    WholeTableNotNullAndNotNaNValidation,
+    ValueSetValidation,
+    SRIDUniqueNonZero
+)
+
 Base = declarative_base()
 
 
@@ -853,4 +862,82 @@ def __init__(self, dependencies):
             version=self.version,
             dependencies=dependencies,
             tasks=tasks,
+            validation={
+                "data-quality": [
+                    RowCountValidation(
+                        table="supply.egon_chp_plants",
+                        rule_id="ROW_COUNT.egon_chp_plants",
+                        expected_count={
+                            "Schleswig-Holstein": 1720,
+                            "Everything": 40197
+                        }
+                    ),
+                    DataTypeValidation(
+                        table="supply.egon_chp_plants",
+                        rule_id="DATA_TYPES.egon_chp_plants",
+                        column_types={
+                            "id": "integer",
+                            "sources": "jsonb",
+                            "source_id": "jsonb",
+                            "carrier": "character varying",
+                            "district_heating": "boolean",
+                            "el_capacity": "double precision",
+                            "th_capacity": "double precision",
+                            "electrical_bus_id": "integer",
+                            "district_heating_area_id": "integer",
+                            "ch4_bus_id": "integer",
+                            "voltage_level": "integer",
+                            "scenario": "character varying",
+                            "geom": "geometry"
+                        }
+                    ),
+                    NotNullAndNotNaNValidation(
+                        table="supply.egon_chp_plants",
+                        rule_id="NOT_NAN.egon_chp_plants",
+                        columns=[
+                            "id",
+                            "sources",
+                            "source_id",
+                            "carrier",
+                            "district_heating",
+                            "el_capacity",
+                            "th_capacity",
+                            "electrical_bus_id",
+                            "district_heating_area_id",
+                            "ch4_bus_id",
+                            "voltage_level",
+                            "scenario",
+                            "geom"
+                        ]
+                    ),
+                    WholeTableNotNullAndNotNaNValidation(
+                        table="supply.egon_chp_plants",
+                        rule_id="TABLE_NOT_NAN.egon_chp_plants"
+                    ),
+                    ValueSetValidation(
+                        table="supply.egon_chp_plants",
+                        rule_id="VALUE_SET_VALIDATION_CARRIER.egon_chp_plants",
+                        column="carrier",
+                        expected_values=[
+                            "oil",
+                            "others",
+                            "gas",
+                            "gas extended",
+                            "biomass"
+                        ]
+                    ),
+                    ValueSetValidation(
+                        table="supply.egon_chp_plants",
+                        rule_id="VALUE_SET_VALIDATION_SCENARIO.egon_chp_plants",
+                        column="scenario",
+                        expected_values=["eGon2035", "eGon100RE"]
+                    ),
+                    SRIDUniqueNonZero(
+                        table="supply.egon_chp_plants",
+                        rule_id="SRIDUniqueNonZero.egon_chp_plants",
+                        column="geom"
+                    )
+                ]
+            },
+            on_validation_failure="continue"
         )

src/egon/data/datasets/demandregio/__init__.py

@@ -20,6 +20,13 @@
 )
 import egon.data.config
 import egon.data.datasets.scenario_parameters.parameters as scenario_parameters
+from egon_validation import (
+    RowCountValidation,
+    DataTypeValidation,
+    WholeTableNotNullAndNotNaNValidation,
+    ValueSetValidation,
+    ArrayCardinalityValidation
+)
 
 try:
     from disaggregator import config, data, spatial, temporal
@@ -87,6 +94,68 @@ def __init__(self, dependencies):
                     insert_cts_ind_demands,
                 },
             ),
+            validation={
+                "data_quality": [
+                    RowCountValidation(
+                        table=" demand.egon_demandregio_hh",
+                        rule_id="ROW_COUNT.egon_demandregio_hh",
+                        expected_count={
+                            "Schleswig-Holstein": 180,
+                            "everything": 7218
+                        }
+                    ),
+                    DataTypeValidation(
+                        table="demand.egon_demandregio_hh",
+                        rule_id="DATA_MULTIPLE_TYPES.egon_demandregio_hh",
+                        column_types={"nuts3": "character varying",
+                                      "hh_size": "integer",
+                                      "scenario": "character varying",
+                                      "year": "integer",
+                                      "demand": "double precision"
+                                      }
+                    ),
+                    WholeTableNotNullAndNotNaNValidation(
+                        table="demand.egon_demandregio_hh",
+                        rule_id="WHOLE_TABLE_NOT_NAN.egon_demandregio_hh"
+                    ),
+                    ValueSetValidation(
+                        table="demand.egon_demandregio_hh",
+                        rule_id="VALUE_SET_VALIDATION_SCENARIO.egon_demandregio_hh",
+                        column="scenario",
+                        expected_values=["eGon2035", "eGon100RE", "eGon2021"]
+                    ),
+                    RowCountValidation(
+                        table=" demand.egon_demandregio_wz",
+                        rule_id="ROW_COUNT.egon_demandregio_wz",
+                        expected_count=87
+                    ),
+                    DataTypeValidation(
+                        table="demand.egon_demandregio_wz",
+                        rule_id="DATA_MULTIPLE_TYPES.egon_demandregio_wz",
+                        column_types={"wz": "integer",
+                                      "sector": "character varying",
+                                      "definition": "character varying"
+                                      }
+                    ),
+                    WholeTableNotNullAndNotNaNValidation(
+                        table="demand.egon_demandregio_wz",
+                        rule_id="WHOLE_TABLE_NOT_NAN.egon_demandregio_wz"
+                    ),
+                    ValueSetValidation(
+                        table="demand.egon_demandregio_wz",
+                        rule_id="VALUE_SET_VALIDATION_SECTOR.egon_demandregio_wz",
+                        column="sector",
+                        expected_values=["industry", "CTS"]
+                    ),
+                    ArrayCardinalityValidation(
+                        table="demand.egon_demandregio_sites_ind_electricity_dsm_timeseries",
+                        rule_id="ARRAY_VALIDATION.egon_demandregio_sites_ind_electricity_dsm_timeseries",
+                        array_column="load_curve",
+                        expected_length=8760,
+                    )
+                ]
+            },
+            on_validation_failure="continue"
         )