Source code for dove.core.components

# Copyright 2024, Battelle Energy Alliance, LLC
# ALL RIGHTS RESERVED
"""
``dove.core.components``
========================

Module containing component classes that form the basis of DOVE models.

Components are connected through resources they consume and produce, forming
a directed graph that represents a system model. Each component can have
associated cash flows to motivate the optimization and transfer functions that define
how resources flow through the component.

Classes
----------
    - `Component`: Abstract base class for all components in a DOVE model
    - `Source`: Component that produces a resource
    - `Sink`: Component that consumes a resource
    - `Converter`: Component that transforms input resources into output resources
    - `Storage`: Component that can store a resource for later use

Types
----------
    - `TimeDependent`: Type alias for time series data (list of floats or NumPy array)
"""

from __future__ import annotations

import warnings
from abc import ABC
from dataclasses import dataclass, field
from typing import Any, Literal, TypeAlias

import numpy as np
from numpy.typing import NDArray

from .cashflow import CashFlow
from .resource import Resource
from .transfers import RatioTransfer, TransferFunc

TimeDependent: TypeAlias = list[float] | NDArray[np.float64]


[docs] @dataclass(kw_only=True) class Component(ABC): """ A base component class for energy system modeling. This abstract class defines the common interface and validation for components in an energy system. Components have time-dependent capacity constraints, can consume and produce resources, and can have associated cashflows. Attributes ---------- name : str Unique identifier for the component. max_capacity_profile : TimeDependent Time-dependent maximum operational capacity of the component. consumes : list[Resource] Resources consumed by this component. produces : list[Resource] Resources produced by this component. min_capacity_profile : TimeDependent Time-dependent minimum operational capacity; defaults to 0.0 for every timestep. capacity_resource : Resource | None The resource that defines capacity, if any. flexibility : Literal["flex", "fixed"] Whether the component's operation is flexible or fixed in optimization. cashflows : list[CashFlow] Economic cashflows associated with this component. transfer_fn : TransferFunc | None Function defining resource conversion logic. Notes ----- This is an abstract base class that should be subclassed to implement specific component types like generators, grids, storage, etc. """ name: str max_capacity_profile: TimeDependent consumes: list[Resource] = field(default_factory=list) produces: list[Resource] = field(default_factory=list) min_capacity_profile: TimeDependent = field(default_factory=list) capacity_resource: Resource | None = None flexibility: Literal["flex", "fixed"] = "flex" cashflows: list[CashFlow] = field(default_factory=list) transfer_fn: TransferFunc | None = None @property def produces_by_name(self) -> list[str]: """A list of resource names that are produced by this component.""" return [r.name for r in self.produces] @property def consumes_by_name(self) -> list[str]: """A list of resource names that are consumed by this component.""" return [r.name for r in self.consumes] def __post_init__(self) -> None: """ Validate and process the component's attributes after initialization. This method performs various checks and conversions to ensure the component is properly configured: 1. Converts max_capacity_profile to a numpy array of floats 2. Validates max_capacity_profile values 3. Checks that flexibility is either 'flex' or 'fixed' 4. Warns if fixed flexibility and min_capacity_profile were both set by the user 5. Sets min_capacity_profile equal to max_capacity_profile if flexibility is fixed 6. Populates min_capacity_profile if necessary 7. Converts min_capacity_profile to a numpy array of floats if necessary 8. Validates min_capacity_profile values 9. Verifies all resources are Resource instances 10. Ensures capacity_resource is in either consumes or produces 11. Calls _validate_cashflows to validate cashflows Raises ------ ValueError If any validation check fails TypeError If resources is not a list of Resource instances UserWarning If min_capacity_profile and fixed flexibility were both explicitly specified for a component """ # convert max_capacity_profile self.max_capacity_profile = np.asarray(self.max_capacity_profile, float).ravel() # validate max_capacity_profile if (self.max_capacity_profile < 0).any(): raise ValueError(f"{self.name}: max_capacity_profile contains negative values") # flexibility if self.flexibility not in ("flex", "fixed"): raise ValueError( f"{self.name}: flexibility must be 'flex' or 'fixed', got {self.flexibility}" ) if self.flexibility == "fixed": if len(self.min_capacity_profile) > 0: warnings.warn( f"{self.name}: both min_capacity_profile and fixed flexibility were specified. " "Overriding min_capacity_profile in order to fix profile at max_capacity_profile", UserWarning, stacklevel=2, ) self.min_capacity_profile = self.max_capacity_profile if len(self.min_capacity_profile) < 1: # set default min_capacity_profile self.min_capacity_profile = np.full(len(self.max_capacity_profile), 0.0) else: # convert min_capacity_profile self.min_capacity_profile = np.asarray(self.min_capacity_profile, float).ravel() # validate min_capacity_profile if len(self.min_capacity_profile) != len(self.max_capacity_profile): raise ValueError( f"{self.name}: length of min_capacity_profile does not equal length of max_capacity_profile" f"({len(self.min_capacity_profile)} != {len(self.max_capacity_profile)})" ) for min_cap_value, max_cap_value in zip( self.min_capacity_profile, self.max_capacity_profile, strict=True ): if not (0 <= min_cap_value <= max_cap_value): raise ValueError( f"{self.name}: each value in min_capacity_profile must be in " f"[0, <max_capacity_profile value>] ({min_cap_value} is not in [0, {max_cap_value}])" ) # resources if not all(isinstance(res, Resource) for res in self.produces + self.consumes): raise TypeError(f"{self.name}: all resources must be Resource instances") # capacity resource consistency if self.capacity_resource: allowed = set(self.consumes + self.produces) if self.capacity_resource not in allowed: raise ValueError( f"{self.name}: capacity_resource {self.capacity_resource} " "not in consumes or produces" ) # cashflows self._validate_cashflows() def _validate_cashflows(self) -> None: """ Validate cashflows and expand cashflow price_profiles if not provided by the user. Raises ------ TypeError: If cashflows contains elements that are not CashFlow instances ValueError: If any cashflow's price_profile is a different length from the component's capacity profile length """ for cf in self.cashflows: if not isinstance(cf, CashFlow): raise TypeError(f"{self.name}: all cashflows must be CashFlow instances") if len(cf.price_profile) < 1: cf.price_profile = np.full(len(self.max_capacity_profile), cf.alpha) elif len(cf.price_profile) != len(self.max_capacity_profile): raise ValueError( f"{self.name}: {cf.name}: cashflow price_profile length " "does not match component profile length!" )
[docs] @dataclass class Source(Component): """ A Source component that produces a specified resource in the DOVE simulation. This component acts as a resource generator or provider in the system simulation, producing a specific resource type with a configurable capacity. Parameters ---------- name : str The unique identifier for this source component. produces : Resource The resource type that this source generates. **kwargs : Any Additional arguments to pass to the parent Component class. Raises ------ ValueError If 'consumes' or 'capacity_resource' is found in the kwargs Notes ----- By default, a Source uses a RatioTransfer function with a 1:1 ratio, meaning it will output the same amount of resource as requested (up to its capacity limit). """
[docs] def __init__(self, name: str, produces: Resource, **kwargs: Any) -> None: if "consumes" in kwargs: raise ValueError( f"{name}: Keyword argument 'consumes' was provided for a Source, but sources cannot consume." "Please remove keyword argument." ) if "capacity_resource" in kwargs: if kwargs["capacity_resource"] == produces: del kwargs["capacity_resource"] else: raise ValueError( f"{name}: Keyword argument 'capacity_resource' was provided as a different resource " "than the component produces, which is not permitted for a Source. Please set " f"'capacity_resource' to {produces} explicitly, or remove it, and it will be set implicitly." ) super().__init__(name=name, produces=[produces], capacity_resource=produces, **kwargs) if self.transfer_fn is None: res = produces self.transfer_fn = RatioTransfer(input_res=res, output_res=res, ratio=1.0)
[docs] @dataclass class Sink(Component): """ A component that consumes a resource and acts as an endpoint in the DOVE system. A Sink is a specialized Component that only consumes resources without producing any. It represents destinations where resources exit the system, such as exporters, consumers, or terminal points in a resource flow network. Parameters ---------- name : str The unique identifier for this sink component. consumes : Resource The resource type this sink consumes. **kwargs : Any Additional keyword arguments to pass to the Component parent class. Raises ------ ValueError If 'produces' or 'capacity_resource' is found in the kwargs Notes ----- By default a Sink creates a `RatioTransfer` function with a 1:1 ratio using the consumed resource as both input and mocked output. """
[docs] def __init__(self, name: str, consumes: Resource, **kwargs: Any) -> None: if "produces" in kwargs: raise ValueError( f"{name}: Keyword argument 'produces' was specified for a Sink, but sinks cannot produce." "Please remove keyword argument." ) if "capacity_resource" in kwargs: if kwargs["capacity_resource"] == consumes: del kwargs["capacity_resource"] else: raise ValueError( f"{name}: Keyword argument 'capacity_resource' was specified as a different resource " "than the component consumes, which is not permitted for a Sink. Please set " f"'capacity_resource' to {consumes} explicitly, or remove it, and it will be set implicitly." ) super().__init__(name=name, consumes=[consumes], capacity_resource=consumes, **kwargs) if self.transfer_fn is None: res = consumes self.transfer_fn = RatioTransfer(input_res=res, output_res=res, ratio=1.0)
[docs] @dataclass class Converter(Component): """ A Component subclass that converts resources from one type to another. The Converter class represents components that consume one set of resources and produce another, potentially different set of resources. Parameters ---------- ramp_limit : float Maximum rate of change for the component's capacity utilization between timesteps as a percent of capacity. Must be a value between 0 and 1. Default is 1.0 (no limit). ramp_freq : int Frequency at which ramp limitations are applied. Default is 0 (every timestep). **kwargs : Any Additional keyword arguments to pass to the parent Component class. Notes ----- - A Converter requires a 'capacity_resource' to determine its capacity limits. - If not specified, the capacity_resource will be automatically determined: - If the component consumes and produces the same resource, that resource is used. - Otherwise, capacity_resource must be explicitly specified. - When resources differ between input and output, a transfer_fn is required to define the conversion relationship. """ ramp_limit: float = 1.0 ramp_freq: int = 0 def __post_init__(self) -> None: """ Perform post-initialization validation and setup. This method calls the parent class's post-initialization and then: 1. Validates ramp_limit and ramp_freq values 2. If no capacity_resource was provided, attempts to determine one: - If the same resource is consumed and produced, uses that resource - Otherwise raises an error indicating ambiguity 3. Warns or raises errors if capacity_resource is ambiguous Raises ------ ValueError If capacity_resource is ambiguous or missing when required UserWarning If capacity_resource was not specified but could be determined automatically """ super().__post_init__() if not (0.0 <= self.ramp_limit <= 1.0): raise ValueError( f"Converter {self.name}: 'ramp_limit'={self.ramp_limit} is outside the range [0, 1].", ) if self.ramp_freq < 0: raise ValueError( f"Converter {self.name}: 'ramp_freq'={self.ramp_freq} must be positive" ) # If no capacity_resource was provided, pick one or error out if self.capacity_resource is None and self.consumes and self.produces: in_res = self.consumes[0] out_res = self.produces[0] if in_res is out_res: # unambiguous: same resource on both sides self.capacity_resource = in_res warnings.warn( f"Converter {self.name}: capacity_resource not specified, " f"using common resource '{in_res.name}'.", UserWarning, stacklevel=2, ) else: # ambiguous capacity_resource if self.transfer_fn is None: raise ValueError( f"Converter {self.name}: ambiguous capacity_resource (consumes {self.consumes_by_name} " f"and produces {self.produces_by_name}) and missing transfer_fn; " "please specify capacity_resource and transfer_fn explicitly." ) raise ValueError( f"Converter {self.name}: ambiguous capacity_resource (consumes {self.consumes_by_name} " f"and produces {self.produces_by_name}); " "please specify capacity_resource explicitly." )
[docs] @dataclass class Storage(Component): """ Storage component for modeling energy storage systems. This class represents a storage component in the energy system, capable of charging and discharging a specific resource. Attributes ---------- resource : Resource The resource type that this storage can store. rte : float Round-trip efficiency, between 0 and 1. Defaults to 1.0. max_charge_rate : float Maximum charge rate as a fraction of capacity per time period, between 0 and 1. Defaults to 1.0. max_discharge_rate : float Maximum discharge rate as a fraction of capacity per time period, between 0 and 1. Defaults to 1.0. initial_stored : float Initial stored amount as a fraction of capacity, between 0 and 1. Defaults to 0.0. periodic_level : bool If True, storage level at the end of the simulation must equal the initial level. Defaults to True. **kwargs : Any Additional keyword arguments to pass to the parent Component class. It's important to note that some Component attributes may not be applicable to Storage components, such as `produces`, `consumes`, and `capacity_resource`. These attributes are set automatically based on the `resource` parameter as necessary. Notes ----- Storage components are treated differently from other components in terms of capacity and resource management. They can store energy for later use, and their operation is governed by the round-trip efficiency (rte) and charge/discharge rates. """ resource: Resource rte: float = 1.0 max_charge_rate: float = 1.0 max_discharge_rate: float = 1.0 initial_stored: float = 0.0 periodic_level: bool = True def __post_init__(self) -> None: """ Initialize the Storage component after instance creation. This method validates input parameters of the Storage component: - Checks that unaccepted attributes for Storage components have not been added - Checks if parameters (rte, max_charge_rate, max_discharge_rate, initial_stored) are within the range [0, 1] - Calls the parent class's post-initialization method for additional validation - Sets the capacity_resource to resource Raises ------ ValueError If 'produces', 'consumes', or 'capacity_resource' was added If 'flexibility' was set to 'fixed' If any of the rate parameters are outside the range [0, 1] If 'max_capacity_profile' is not constant """ # Error if unaccepted attribute was added for bad_attr in ("produces", "consumes", "capacity_resource"): if getattr(self, bad_attr): raise ValueError( f"Unaccepted keyword argument '{bad_attr}' provided to Storage {self.name}. " "Please remove keyword argument." ) # Error if flexibility was set to fixed if self.flexibility == "fixed": raise ValueError( f"Attribute 'flexibility' was set to 'fixed' on Storage {self.name}. " "Storage components must be flexible." ) # Error if parameters are outside [0, 1] for attr in ("rte", "max_charge_rate", "max_discharge_rate", "initial_stored"): val = getattr(self, attr) if not (0.0 <= val <= 1.0): raise ValueError( f"Storage {self.name}: '{attr}'={val} is outside the range [0, 1].", ) super().__post_init__() # Error if max_capacity_profile is not constant if not np.all(self.max_capacity_profile == self.max_capacity_profile[0]): raise ValueError( f"Non-constant max_capacity_profile was added to storage {self.name}. " "Storage components must have constant max_capacity_profile values." ) # Set capacity_resource self.capacity_resource = self.resource