2024-12-15 14:40:03 +01:00
|
|
|
from typing import Optional
|
|
|
|
|
2024-04-02 16:46:16 +02:00
|
|
|
import numpy as np
|
2024-11-15 22:27:25 +01:00
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
2025-01-05 14:41:07 +01:00
|
|
|
from akkudoktoreos.core.logging import get_logger
|
2024-12-15 14:40:03 +01:00
|
|
|
from akkudoktoreos.devices.devicesabc import DeviceBase
|
|
|
|
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
2024-11-15 22:27:25 +01:00
|
|
|
|
2024-11-26 00:53:16 +01:00
|
|
|
class HomeApplianceParameters(BaseModel):
|
|
|
|
consumption_wh: int = Field(
|
2024-11-15 22:27:25 +01:00
|
|
|
gt=0,
|
|
|
|
description="An integer representing the energy consumption of a household device in watt-hours.",
|
|
|
|
)
|
2024-11-26 00:53:16 +01:00
|
|
|
duration_h: int = Field(
|
2024-11-15 22:27:25 +01:00
|
|
|
gt=0,
|
|
|
|
description="An integer representing the usage duration of a household device in hours.",
|
|
|
|
)
|
2024-04-02 16:46:16 +02:00
|
|
|
|
2024-10-03 11:05:44 +02:00
|
|
|
|
2024-12-15 14:40:03 +01:00
|
|
|
class HomeAppliance(DeviceBase):
|
|
|
|
def __init__(
|
|
|
|
self,
|
|
|
|
parameters: Optional[HomeApplianceParameters] = None,
|
|
|
|
hours: Optional[int] = 24,
|
|
|
|
provider_id: Optional[str] = None,
|
|
|
|
):
|
|
|
|
# Configuration initialisation
|
|
|
|
self.provider_id = provider_id
|
|
|
|
self.prefix = "<invalid>"
|
|
|
|
if self.provider_id == "GenericDishWasher":
|
|
|
|
self.prefix = "dishwasher"
|
|
|
|
# Parameter initialisiation
|
|
|
|
self.parameters = parameters
|
|
|
|
if hours is None:
|
|
|
|
self.hours = self.total_hours
|
|
|
|
else:
|
|
|
|
self.hours = hours
|
|
|
|
|
|
|
|
self.initialised = False
|
|
|
|
# Run setup if parameters are given, otherwise setup() has to be called later when the config is initialised.
|
|
|
|
if self.parameters is not None:
|
|
|
|
self.setup()
|
|
|
|
|
|
|
|
def setup(self) -> None:
|
|
|
|
if self.initialised:
|
|
|
|
return
|
|
|
|
if self.provider_id is not None:
|
|
|
|
# Setup by configuration
|
|
|
|
self.hours = self.total_hours
|
|
|
|
self.consumption_wh = getattr(self.config, f"{self.prefix}_consumption")
|
|
|
|
self.duration_h = getattr(self.config, f"{self.prefix}_duration")
|
|
|
|
elif self.parameters is not None:
|
|
|
|
# Setup by parameters
|
|
|
|
self.consumption_wh = (
|
|
|
|
self.parameters.consumption_wh
|
|
|
|
) # Total energy consumption of the device in kWh
|
|
|
|
self.duration_h = self.parameters.duration_h # Duration of use in hours
|
|
|
|
else:
|
|
|
|
error_msg = "Parameters and provider ID missing. Can't instantiate."
|
|
|
|
logger.error(error_msg)
|
|
|
|
raise ValueError(error_msg)
|
2024-11-26 00:53:16 +01:00
|
|
|
self.load_curve = np.zeros(self.hours) # Initialize the load curve with zeros
|
2024-12-15 14:40:03 +01:00
|
|
|
self.initialised = True
|
2024-04-02 16:46:16 +02:00
|
|
|
|
2024-11-26 22:28:05 +01:00
|
|
|
def set_starting_time(self, start_hour: int, global_start_hour: int = 0) -> None:
|
2024-11-10 23:27:52 +01:00
|
|
|
"""Sets the start time of the device and generates the corresponding load curve.
|
|
|
|
|
2024-09-20 12:23:22 +02:00
|
|
|
:param start_hour: The hour at which the device should start.
|
2024-04-02 16:46:16 +02:00
|
|
|
"""
|
2024-12-15 14:40:03 +01:00
|
|
|
self.reset_load_curve()
|
2024-09-20 12:23:22 +02:00
|
|
|
# Check if the duration of use is within the available time frame
|
2024-11-26 00:53:16 +01:00
|
|
|
if start_hour + self.duration_h > self.hours:
|
2024-09-20 12:23:22 +02:00
|
|
|
raise ValueError("The duration of use exceeds the available time frame.")
|
2024-04-02 16:46:16 +02:00
|
|
|
if start_hour < global_start_hour:
|
2024-09-20 12:23:22 +02:00
|
|
|
raise ValueError("The start time is earlier than the available time frame.")
|
2024-10-03 11:05:44 +02:00
|
|
|
|
2024-09-20 12:23:22 +02:00
|
|
|
# Calculate power per hour based on total consumption and duration
|
2024-11-26 00:53:16 +01:00
|
|
|
power_per_hour = self.consumption_wh / self.duration_h # Convert to watt-hours
|
2024-10-03 11:05:44 +02:00
|
|
|
|
2024-09-20 12:23:22 +02:00
|
|
|
# Set the power for the duration of use in the load curve array
|
2024-11-26 00:53:16 +01:00
|
|
|
self.load_curve[start_hour : start_hour + self.duration_h] = power_per_hour
|
2024-04-02 16:46:16 +02:00
|
|
|
|
2024-12-15 14:40:03 +01:00
|
|
|
def reset_load_curve(self) -> None:
|
2024-11-10 23:27:52 +01:00
|
|
|
"""Resets the load curve."""
|
2024-11-26 00:53:16 +01:00
|
|
|
self.load_curve = np.zeros(self.hours)
|
2024-04-02 16:46:16 +02:00
|
|
|
|
2024-11-26 22:28:05 +01:00
|
|
|
def get_load_curve(self) -> np.ndarray:
|
2024-11-10 23:27:52 +01:00
|
|
|
"""Returns the current load curve."""
|
2024-11-26 00:53:16 +01:00
|
|
|
return self.load_curve
|
2024-04-02 16:46:16 +02:00
|
|
|
|
2024-11-26 22:28:05 +01:00
|
|
|
def get_load_for_hour(self, hour: int) -> float:
|
2024-11-10 23:27:52 +01:00
|
|
|
"""Returns the load for a specific hour.
|
|
|
|
|
2024-09-20 12:23:22 +02:00
|
|
|
:param hour: The hour for which the load is queried.
|
|
|
|
:return: The load in watts for the specified hour.
|
2024-04-02 16:46:16 +02:00
|
|
|
"""
|
|
|
|
if hour < 0 or hour >= self.hours:
|
2024-09-20 12:23:22 +02:00
|
|
|
raise ValueError("The specified hour is outside the available time frame.")
|
2024-10-03 11:05:44 +02:00
|
|
|
|
2024-11-26 00:53:16 +01:00
|
|
|
return self.load_curve[hour]
|
2024-04-02 16:46:16 +02:00
|
|
|
|
2024-11-26 22:28:05 +01:00
|
|
|
def get_latest_starting_point(self) -> int:
|
2024-11-10 23:27:52 +01:00
|
|
|
"""Returns the latest possible start time at which the device can still run completely."""
|
2024-11-26 00:53:16 +01:00
|
|
|
return self.hours - self.duration_h
|