mirror of
				https://github.com/Akkudoktor-EOS/EOS.git
				synced 2025-11-04 08:46:20 +00:00 
			
		
		
		
	Tests for class_load_container v2 (#247)
* Tests for class_load_container v2 - file as same name as class - useless numpy conversations removed - switched to built-in type tests - human readable tests included * ruff * removed dupicate empty list * load_aggregator: collections.abc.Sequence --------- Co-authored-by: Dominique Lasserre <lasserre.d@gmail.com>
This commit is contained in:
		
							
								
								
									
										37
									
								
								src/akkudoktoreos/prediction/load_aggregator.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										37
									
								
								src/akkudoktoreos/prediction/load_aggregator.py
									
									
									
									
									
										Normal file
									
								
							@@ -0,0 +1,37 @@
 | 
			
		||||
from collections import defaultdict
 | 
			
		||||
from collections.abc import Sequence
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class LoadAggregator:
 | 
			
		||||
    def __init__(self, prediction_hours: int = 24) -> None:
 | 
			
		||||
        """Initializes the LoadAggregator object with the number of prediction hours.
 | 
			
		||||
 | 
			
		||||
        :param prediction_hours: Number of hours to predict (default: 24)
 | 
			
		||||
        """
 | 
			
		||||
        self.loads: defaultdict[str, list[float]] = defaultdict(
 | 
			
		||||
            list
 | 
			
		||||
        )  # Dictionary to hold load arrays for different sources
 | 
			
		||||
        self.prediction_hours: int = prediction_hours
 | 
			
		||||
 | 
			
		||||
    def add_load(self, name: str, last_array: Sequence[float]) -> None:
 | 
			
		||||
        """Adds a load array for a specific source. Accepts a Sequence of floats.
 | 
			
		||||
 | 
			
		||||
        :param name: Name of the load source (e.g., "Household", "Heat Pump").
 | 
			
		||||
        :param last_array: Sequence of loads, where each entry corresponds to an hour.
 | 
			
		||||
        :raises ValueError: If the length of last_array doesn't match the prediction hours.
 | 
			
		||||
        """
 | 
			
		||||
        # Check length of the array without converting
 | 
			
		||||
        if len(last_array) != self.prediction_hours:
 | 
			
		||||
            raise ValueError(f"Total load inconsistent lengths in arrays: {name} {len(last_array)}")
 | 
			
		||||
        self.loads[name] = list(last_array)
 | 
			
		||||
 | 
			
		||||
    def calculate_total_load(self) -> list[float]:
 | 
			
		||||
        """Calculates the total load for each hour by summing up the loads from all sources.
 | 
			
		||||
 | 
			
		||||
        :return: A list representing the total load for each hour.
 | 
			
		||||
                 Returns an empty list if no loads have been added.
 | 
			
		||||
        """
 | 
			
		||||
        # Optimize the summation using a single loop with zip
 | 
			
		||||
        total_load = [sum(hourly_loads) for hourly_loads in zip(*self.loads.values())]
 | 
			
		||||
 | 
			
		||||
        return total_load
 | 
			
		||||
@@ -1,39 +0,0 @@
 | 
			
		||||
import numpy as np
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class Gesamtlast:
 | 
			
		||||
    def __init__(self, prediction_hours: int = 24):
 | 
			
		||||
        self.lasten: dict[
 | 
			
		||||
            str, np.ndarray
 | 
			
		||||
        ] = {}  # Contains names and load arrays for different sources
 | 
			
		||||
        self.prediction_hours = prediction_hours
 | 
			
		||||
 | 
			
		||||
    def hinzufuegen(self, name: str, last_array: np.ndarray) -> None:
 | 
			
		||||
        """Adds an array of loads for a specific source.
 | 
			
		||||
 | 
			
		||||
        :param name: Name of the load source (e.g., "Household", "Heat Pump")
 | 
			
		||||
        :param last_array: Array of loads, where each entry corresponds to an hour
 | 
			
		||||
        """
 | 
			
		||||
        if len(last_array) != self.prediction_hours:
 | 
			
		||||
            raise ValueError(f"Total load inconsistent lengths in arrays: {name} {len(last_array)}")
 | 
			
		||||
        self.lasten[name] = last_array
 | 
			
		||||
 | 
			
		||||
    def gesamtlast_berechnen(self) -> np.ndarray:
 | 
			
		||||
        """Calculates the total load for each hour and returns an array of total loads.
 | 
			
		||||
 | 
			
		||||
        :return: Array of total loads, where each entry corresponds to an hour
 | 
			
		||||
        """
 | 
			
		||||
        if not self.lasten:
 | 
			
		||||
            return np.ndarray(0)
 | 
			
		||||
 | 
			
		||||
        # Assumption: All load arrays have the same length
 | 
			
		||||
        stunden = len(next(iter(self.lasten.values())))
 | 
			
		||||
        gesamtlast_array = [0] * stunden
 | 
			
		||||
 | 
			
		||||
        for last_array in self.lasten.values():
 | 
			
		||||
            gesamtlast_array = [
 | 
			
		||||
                gesamtlast + stundenlast
 | 
			
		||||
                for gesamtlast, stundenlast in zip(gesamtlast_array, last_array)
 | 
			
		||||
            ]
 | 
			
		||||
 | 
			
		||||
        return np.array(gesamtlast_array)
 | 
			
		||||
		Reference in New Issue
	
	Block a user