Code Duplication - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 93-93 lines in 3 locations

src/egon/data/validation/rules/custom/sanity/gas_grid.py 1 location


        except Exception as e:
            raise ValueError(f"Error reading SciGRID_gas reference data: {str(e)}")

    def evaluate_df(self, df, ctx):
        """
        Evaluate CH4 grid capacity against reference data.

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

        Returns
        -------
        RuleResult
            Validation result with success/failure status
        """
        if df.empty or df["total_p_nom"].isna().all():
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                message=f"No CH4 links found for scenario {self.scenario}",
                severity=Severity.WARNING,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        observed_capacity = float(df["total_p_nom"].values[0])

        # Get expected capacity from SciGRID_gas data
        try:
            expected_capacity = self._get_reference_capacity()
        except Exception as e:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                message=str(e),
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        # Calculate relative deviation
        rtol = self.params.get("rtol", 0.10)
        deviation = abs(observed_capacity - expected_capacity) / expected_capacity

        success = deviation <= rtol
        deviation_pct = deviation * 100

        if success:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=True,
                observed=observed_capacity,
                expected=expected_capacity,
                message=(
                    f"CH4 grid capacity valid for {self.scenario}: "
                    f"{observed_capacity:.2f} GWh/d (deviation: {deviation_pct:.2f}%, "
                    f"tolerance: {rtol*100:.2f}%)"
                ),
                severity=Severity.INFO,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )
        else:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                observed=observed_capacity,
                expected=expected_capacity,
                message=(
                    f"CH4 grid capacity deviation too large for {self.scenario}: "
                    f"{observed_capacity:.2f} vs {expected_capacity:.2f} GWh/d expected "
                    f"(deviation: {deviation_pct:.2f}%, tolerance: {rtol*100:.2f}%)"
                ),
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )



src/egon/data/validation/rules/custom/sanity/gas_loads_generators.py 2 locations


        except Exception as e:
            raise ValueError(f"Error reading reference generation data: {str(e)}")

    def evaluate_df(self, df, ctx):
        """
        Evaluate gas generators capacity against reference data.

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

        Returns
        -------
        RuleResult
            Validation result with success/failure status
        """
        if df.empty or df["p_nom_germany"].isna().all():
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                message=f"No {self.carrier} generators found for scenario {self.scenario}",
                severity=Severity.WARNING,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        observed_capacity = float(df["p_nom_germany"].values[0])

        # Get expected capacity from reference data
        try:
            expected_capacity = self._get_reference_capacity()
        except Exception as e:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                message=str(e),
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        # Calculate relative deviation
        rtol = self.params.get("rtol", 0.10)
        deviation = abs(observed_capacity - expected_capacity) / expected_capacity

        success = deviation <= rtol
        deviation_pct = deviation * 100

        if success:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=True,
                observed=observed_capacity,
                expected=expected_capacity,
                message=(
                    f"{self.carrier} generator capacity valid for {self.scenario}: "
                    f"{observed_capacity:.2f} MW (deviation: {deviation_pct:.2f}%, "
                    f"tolerance: {rtol*100:.2f}%)"
                ),
                severity=Severity.INFO,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )
        else:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                observed=observed_capacity,
                expected=expected_capacity,
                message=(
                    f"{self.carrier} generator capacity deviation too large for {self.scenario}: "
                    f"{observed_capacity:.2f} vs {expected_capacity:.2f} MW expected "
                    f"(deviation: {deviation_pct:.2f}%, tolerance: {rtol*100:.2f}%)"
                ),
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )


        except Exception as e:
            raise ValueError(f"Error reading reference load data: {str(e)}")

    def evaluate_df(self, df, ctx):
        """
        Evaluate gas loads capacity against reference data.

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

        Returns
        -------
        RuleResult
            Validation result with success/failure status
        """
        if df.empty or df["load_twh"].isna().all():
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                message=f"No {self.carrier} loads found for scenario {self.scenario}",
                severity=Severity.WARNING,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        observed_load = float(df["load_twh"].values[0])

        # Get expected capacity from reference data
        try:
            expected_load = self._get_reference_capacity()
        except Exception as e:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                message=str(e),
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )

        # Calculate relative deviation
        rtol = self.params.get("rtol", 0.10)
        deviation = abs(observed_load - expected_load) / expected_load

        success = deviation <= rtol
        deviation_pct = deviation * 100

        if success:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=True,
                observed=observed_load,
                expected=expected_load,
                message=(
                    f"{self.carrier} load valid for {self.scenario}: "
                    f"{observed_load:.2f} TWh (deviation: {deviation_pct:.2f}%, "
                    f"tolerance: {rtol*100:.2f}%)"
                ),
                severity=Severity.INFO,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )
        else:
            return RuleResult(
                rule_id=self.rule_id,
                task=self.task,
                table=self.table,
                kind=self.kind,
                success=False,
                observed=observed_load,
                expected=expected_load,
                message=(
                    f"{self.carrier} load deviation too large for {self.scenario}: "
                    f"{observed_load:.2f} vs {expected_load:.2f} TWh expected "
                    f"(deviation: {deviation_pct:.2f}%, tolerance: {rtol*100:.2f}%)"
                ),
                severity=Severity.ERROR,
                schema=self.schema,
                table_name=self.table_name,
                rule_class=self.__class__.__name__
            )



		@@ 597-689 (lines=93) @@
594		except Exception as e:
595		raise ValueError(f"Error reading SciGRID_gas reference data: {str(e)}")
596
597		def evaluate_df(self, df, ctx):
598		"""
599		Evaluate CH4 grid capacity against reference data.
600
601		Parameters
602		----------
603		df : pd.DataFrame
604		DataFrame with total_p_nom column
605		ctx : dict
606		Context information
607
608		Returns
609		-------
610		RuleResult
611		Validation result with success/failure status
612		"""
613		if df.empty or df["total_p_nom"].isna().all():
614		return RuleResult(
615		rule_id=self.rule_id,
616		task=self.task,
617		table=self.table,
618		kind=self.kind,
619		success=False,
620		message=f"No CH4 links found for scenario {self.scenario}",
621		severity=Severity.WARNING,
622		schema=self.schema,
623		table_name=self.table_name,
624		rule_class=self.__class__.__name__
625		)
626
627		observed_capacity = float(df["total_p_nom"].values[0])
628
629		# Get expected capacity from SciGRID_gas data
630		try:
631		expected_capacity = self._get_reference_capacity()
632		except Exception as e:
633		return RuleResult(
634		rule_id=self.rule_id,
635		task=self.task,
636		table=self.table,
637		kind=self.kind,
638		success=False,
639		message=str(e),
640		severity=Severity.ERROR,
641		schema=self.schema,
642		table_name=self.table_name,
643		rule_class=self.__class__.__name__
644		)
645
646		# Calculate relative deviation
647		rtol = self.params.get("rtol", 0.10)
648		deviation = abs(observed_capacity - expected_capacity) / expected_capacity
649
650		success = deviation <= rtol
651		deviation_pct = deviation * 100
652
653		if success:
654		return RuleResult(
655		rule_id=self.rule_id,
656		task=self.task,
657		table=self.table,
658		kind=self.kind,
659		success=True,
660		observed=observed_capacity,
661		expected=expected_capacity,
662		message=(
663		f"CH4 grid capacity valid for {self.scenario}: "
664		f"{observed_capacity:.2f} GWh/d (deviation: {deviation_pct:.2f}%, "
665		f"tolerance: {rtol*100:.2f}%)"
666		),
667		severity=Severity.INFO,
668		schema=self.schema,
669		table_name=self.table_name,
670		rule_class=self.__class__.__name__
671		)
672		else:
673		return RuleResult(
674		rule_id=self.rule_id,
675		task=self.task,
676		table=self.table,
677		kind=self.kind,
678		success=False,
679		observed=observed_capacity,
680		expected=expected_capacity,
681		message=(
682		f"CH4 grid capacity deviation too large for {self.scenario}: "
683		f"{observed_capacity:.2f} vs {expected_capacity:.2f} GWh/d expected "
684		f"(deviation: {deviation_pct:.2f}%, tolerance: {rtol*100:.2f}%)"
685		),
686		severity=Severity.ERROR,
687		schema=self.schema,
688		table_name=self.table_name,
689		rule_class=self.__class__.__name__
690		)
691
692

		@@ 319-411 (lines=93) @@
316		except Exception as e:
317		raise ValueError(f"Error reading reference generation data: {str(e)}")
318
319		def evaluate_df(self, df, ctx):
320		"""
321		Evaluate gas generators capacity against reference data.
322
323		Parameters
324		----------
325		df : pd.DataFrame
326		DataFrame with p_nom_germany column
327		ctx : dict
328		Context information
329
330		Returns
331		-------
332		RuleResult
333		Validation result with success/failure status
334		"""
335		if df.empty or df["p_nom_germany"].isna().all():
336		return RuleResult(
337		rule_id=self.rule_id,
338		task=self.task,
339		table=self.table,
340		kind=self.kind,
341		success=False,
342		message=f"No {self.carrier} generators found for scenario {self.scenario}",
343		severity=Severity.WARNING,
344		schema=self.schema,
345		table_name=self.table_name,
346		rule_class=self.__class__.__name__
347		)
348
349		observed_capacity = float(df["p_nom_germany"].values[0])
350
351		# Get expected capacity from reference data
352		try:
353		expected_capacity = self._get_reference_capacity()
354		except Exception as e:
355		return RuleResult(
356		rule_id=self.rule_id,
357		task=self.task,
358		table=self.table,
359		kind=self.kind,
360		success=False,
361		message=str(e),
362		severity=Severity.ERROR,
363		schema=self.schema,
364		table_name=self.table_name,
365		rule_class=self.__class__.__name__
366		)
367
368		# Calculate relative deviation
369		rtol = self.params.get("rtol", 0.10)
370		deviation = abs(observed_capacity - expected_capacity) / expected_capacity
371
372		success = deviation <= rtol
373		deviation_pct = deviation * 100
374
375		if success:
376		return RuleResult(
377		rule_id=self.rule_id,
378		task=self.task,
379		table=self.table,
380		kind=self.kind,
381		success=True,
382		observed=observed_capacity,
383		expected=expected_capacity,
384		message=(
385		f"{self.carrier} generator capacity valid for {self.scenario}: "
386		f"{observed_capacity:.2f} MW (deviation: {deviation_pct:.2f}%, "
387		f"tolerance: {rtol*100:.2f}%)"
388		),
389		severity=Severity.INFO,
390		schema=self.schema,
391		table_name=self.table_name,
392		rule_class=self.__class__.__name__
393		)
394		else:
395		return RuleResult(
396		rule_id=self.rule_id,
397		task=self.task,
398		table=self.table,
399		kind=self.kind,
400		success=False,
401		observed=observed_capacity,
402		expected=expected_capacity,
403		message=(
404		f"{self.carrier} generator capacity deviation too large for {self.scenario}: "
405		f"{observed_capacity:.2f} vs {expected_capacity:.2f} MW expected "
406		f"(deviation: {deviation_pct:.2f}%, tolerance: {rtol*100:.2f}%)"
407		),
408		severity=Severity.ERROR,
409		schema=self.schema,
410		table_name=self.table_name,
411		rule_class=self.__class__.__name__
412		)
413
		@@ 110-202 (lines=93) @@
107		except Exception as e:
108		raise ValueError(f"Error reading reference load data: {str(e)}")
109
110		def evaluate_df(self, df, ctx):
111		"""
112		Evaluate gas loads capacity against reference data.
113
114		Parameters
115		----------
116		df : pd.DataFrame
117		DataFrame with load_twh column
118		ctx : dict
119		Context information
120
121		Returns
122		-------
123		RuleResult
124		Validation result with success/failure status
125		"""
126		if df.empty or df["load_twh"].isna().all():
127		return RuleResult(
128		rule_id=self.rule_id,
129		task=self.task,
130		table=self.table,
131		kind=self.kind,
132		success=False,
133		message=f"No {self.carrier} loads found for scenario {self.scenario}",
134		severity=Severity.WARNING,
135		schema=self.schema,
136		table_name=self.table_name,
137		rule_class=self.__class__.__name__
138		)
139
140		observed_load = float(df["load_twh"].values[0])
141
142		# Get expected capacity from reference data
143		try:
144		expected_load = self._get_reference_capacity()
145		except Exception as e:
146		return RuleResult(
147		rule_id=self.rule_id,
148		task=self.task,
149		table=self.table,
150		kind=self.kind,
151		success=False,
152		message=str(e),
153		severity=Severity.ERROR,
154		schema=self.schema,
155		table_name=self.table_name,
156		rule_class=self.__class__.__name__
157		)
158
159		# Calculate relative deviation
160		rtol = self.params.get("rtol", 0.10)
161		deviation = abs(observed_load - expected_load) / expected_load
162
163		success = deviation <= rtol
164		deviation_pct = deviation * 100
165
166		if success:
167		return RuleResult(
168		rule_id=self.rule_id,
169		task=self.task,
170		table=self.table,
171		kind=self.kind,
172		success=True,
173		observed=observed_load,
174		expected=expected_load,
175		message=(
176		f"{self.carrier} load valid for {self.scenario}: "
177		f"{observed_load:.2f} TWh (deviation: {deviation_pct:.2f}%, "
178		f"tolerance: {rtol*100:.2f}%)"
179		),
180		severity=Severity.INFO,
181		schema=self.schema,
182		table_name=self.table_name,
183		rule_class=self.__class__.__name__
184		)
185		else:
186		return RuleResult(
187		rule_id=self.rule_id,
188		task=self.task,
189		table=self.table,
190		kind=self.kind,
191		success=False,
192		observed=observed_load,
193		expected=expected_load,
194		message=(
195		f"{self.carrier} load deviation too large for {self.scenario}: "
196		f"{observed_load:.2f} vs {expected_load:.2f} TWh expected "
197		f"(deviation: {deviation_pct:.2f}%, tolerance: {rtol*100:.2f}%)"
198		),
199		severity=Severity.ERROR,
200		schema=self.schema,
201		table_name=self.table_name,
202		rule_class=self.__class__.__name__
203		)
204
205

openego / eGon-data

Code Duplication Length = 93-93 lines in 3 locations

src/egon/data/validation/rules/custom/sanity/gas_grid.py 1 location

src/egon/data/validation/rules/custom/sanity/gas_loads_generators.py 2 locations