custom.sanity.electrical_load_sectors.ElectricalLoadSectorBreakdown.evaluate_df()

Complexity

Conditions

6

Size

Total Lines	138
Code Lines	85

Duplication

Lines	12
Ratio	8.7 %

Importance

Changes

0

"""
    def evaluate_df(self, df, ctx):
        """
        Evaluate electrical load sector breakdown.

        Parameters
        ----------
        df : pd.DataFrame
            DataFrame with total load_twh column
        ctx : dict
            Context information

        Returns
        -------
        RuleResult
            Validation result with success/failure status
        """
        if df.empty or df["load_twh"].isna().all():

This code seems to be duplicated in your project.

            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                message=f"No electrical load data found for scenario {self.scenario}",
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        # Get total AC load
        total_load_twh = float(df["load_twh"].values[0])

        # Get sector loads
        try:
            sector_loads = self._get_sector_loads()
        except Exception as e:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                message=f"Error reading sector load data: {str(e)}",
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        # Expected values (from original sanity_checks.py lines 2689-2694)
        # References:
        # https://github.com/openego/powerd-data/blob/56b8215928a8dc4fe953d266c563ce0ed98e93f9/src/egon/data/datasets/demandregio/__init__.py#L480
        # https://github.com/openego/powerd-data/blob/56b8215928a8dc4fe953d266c563ce0ed98e93f9/src/egon/data/datasets/demandregio/__init__.py#L775
        expected_values = {
            "residential": 90.4,
            "commercial": 146.7,
            "industrial": 382.9,
            "total": 620.0
        }

        # Build load summary dataframe
        load_summary = pd.DataFrame({
            "sector": ["residential", "commercial", "industrial", "total"],
            "expected": [
                expected_values["residential"],
                expected_values["commercial"],
                expected_values["industrial"],
                expected_values["total"]
            ],
            "observed": [
                sector_loads["residential"],
                sector_loads["commercial"],
                sector_loads["industrial"],
                total_load_twh
            ]
        })

        load_summary["diff"] = load_summary["observed"] - load_summary["expected"]
        load_summary["diff_pct"] = (
            load_summary["diff"] / load_summary["observed"] * 100
        )

        # Check if all deviations are within tolerance (< 1% as in original)
        violations = load_summary[load_summary["diff_pct"].abs() >= (self.rtol * 100)]

        if not violations.empty:
            # Format violation details
            violation_details = []
            for _, row in violations.iterrows():
                violation_details.append(
                    f"{row['sector']}: {row['observed']:.2f} TWh "
                    f"(expected {row['expected']:.2f} TWh, "
                    f"deviation {row['diff_pct']:+.2f}%)"
                )

            max_deviation = load_summary["diff_pct"].abs().max()

            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                observed=float(max_deviation),
                expected=self.rtol * 100,
                message=(
                    f"Electrical load sector breakdown deviations exceed tolerance for {self.scenario}: "
                    f"{'; '.join(violation_details)}"
                ),
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        # All sectors within tolerance
        sector_summary = "; ".join([
            f"{row['sector']}: {row['observed']:.2f} TWh "
            f"(expected {row['expected']:.2f} TWh, "
            f"deviation {row['diff_pct']:+.2f}%)"
            for _, row in load_summary.iterrows()
        ])

        return RuleResult(
            rule_id=self.rule_id,
            task=self.task,
            table=self.table,
            kind=self.kind,
            success=True,
            observed=0.0,
            expected=0.0,
            message=(
                f"Electrical load sector breakdown valid for {self.scenario}: {sector_summary}"
            ),
            schema=self.schema,
            table_name=self.table_name,
            rule_class=self.__class__.__name__
        )