diff --git a/CITATION.cff b/CITATION.cff index 955db078c0..6dd4f7c413 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -80,6 +80,10 @@ authors: given-names: S. affiliation: United Kingdom Atomic Energy Authority, Culham Science Centre, Abingdon, Oxfordshire OX14 3DB, United Kingdom + - family-names: Pomella Lobo + given-names: T. + affiliation: Karlsruher Institut für Technologie, Hermann-von-Helmholtz-Platz 1, 76344 Eggenstein-Leopoldshafen, Germany + - family-names: Saunders given-names: H. affiliation: United Kingdom Atomic Energy Authority, Culham Science Centre, Abingdon, Oxfordshire OX14 3DB, United Kingdom diff --git a/bluemira/base/constants.py b/bluemira/base/constants.py index d77761b0f6..565031c33c 100644 --- a/bluemira/base/constants.py +++ b/bluemira/base/constants.py @@ -79,6 +79,8 @@ def __init__(self): self.define("atomic_parts_per_million = appm = ppm") # Other currencies need to be set up in a new context self.define("USD = [currency]") + # EU directive volt-ampere reactive is lower case + self.define("var = VA") self._gas_flow_temperature = None self._contexts_added = False diff --git a/bluemira/base/reactor_config.py b/bluemira/base/reactor_config.py index 336cdeb029..5bf8591fa4 100644 --- a/bluemira/base/reactor_config.py +++ b/bluemira/base/reactor_config.py @@ -9,7 +9,7 @@ import pprint from dataclasses import dataclass from pathlib import Path -from typing import Any, Dict, Iterable, Tuple, Type, TypeVar, Union +from typing import Any, Dict, Iterable, Optional, Tuple, Type, TypeVar, Union from bluemira.base.error import ReactorConfigError from bluemira.base.look_and_feel import bluemira_debug, bluemira_warn @@ -94,7 +94,7 @@ class GlobalParams(ParameterFrame): def __init__( self, config_path: Union[str, Path, dict], - global_params_type: Type[_PfT], + global_params_type: Optional[Type[_PfT]] = None, warn_on_duplicate_keys: bool = False, warn_on_empty_local_params: bool = False, warn_on_empty_config: bool = False, @@ -109,7 +109,9 @@ def __init__( self.config_data = config_data self.global_params = make_parameter_frame( - self.config_data.get(_PARAMETERS_KEY, {}), + self.config_data.get( + _PARAMETERS_KEY, None if global_params_type is None else {} + ), global_params_type, ) diff --git a/bluemira/power_cycle/__init__.py b/bluemira/power_cycle/__init__.py new file mode 100644 index 0000000000..228777867c --- /dev/null +++ b/bluemira/power_cycle/__init__.py @@ -0,0 +1,9 @@ +# SPDX-FileCopyrightText: 2021-present M. Coleman, J. Cook, F. Franza +# SPDX-FileCopyrightText: 2021-present I.A. Maione, S. McIntosh +# SPDX-FileCopyrightText: 2021-present J. Morris, D. Short +# +# SPDX-License-Identifier: LGPL-2.1-or-later + +""" +Power cycle module. +""" diff --git a/bluemira/power_cycle/errors.py b/bluemira/power_cycle/errors.py new file mode 100644 index 0000000000..af3bcb860d --- /dev/null +++ b/bluemira/power_cycle/errors.py @@ -0,0 +1,25 @@ +# SPDX-FileCopyrightText: 2021-present M. Coleman, J. Cook, F. Franza +# SPDX-FileCopyrightText: 2021-present I.A. Maione, S. McIntosh +# SPDX-FileCopyrightText: 2021-present J. Morris, D. Short +# +# SPDX-License-Identifier: LGPL-2.1-or-later + +""" +Exception classes for the power cycle model. +""" + +from bluemira.base.error import BluemiraError + + +class PowerCycleError(BluemiraError): + """PowerCycle base error.""" + + +class PowerLoadError(PowerCycleError): + """PowerCycleLoad error class.""" + + +class ScenarioLoadError(PowerCycleError): + """ + Exception class for 'ScenarioLoad' class of the Power Cycle module. + """ diff --git a/bluemira/power_cycle/net.py b/bluemira/power_cycle/net.py new file mode 100644 index 0000000000..1b88ff9586 --- /dev/null +++ b/bluemira/power_cycle/net.py @@ -0,0 +1,865 @@ +# SPDX-FileCopyrightText: 2021-present M. Coleman, J. Cook, F. Franza +# SPDX-FileCopyrightText: 2021-present I.A. Maione, S. McIntosh +# SPDX-FileCopyrightText: 2021-present J. Morris, D. Short +# +# SPDX-License-Identifier: LGPL-2.1-or-later +"""Power cycle net loads""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + Iterable, + List, + Optional, + Set, + Tuple, + Type, + TypedDict, + Union, +) + +import numpy as np +from scipy.interpolate import interp1d +from typing_extensions import NotRequired + +from bluemira.base.constants import raw_uc +from bluemira.base.look_and_feel import bluemira_debug +from bluemira.power_cycle.tools import read_json + +if TYPE_CHECKING: + from pathlib import Path + + import numpy.typing as npt + + +@dataclass +class Config: + """Configuration base dataclass""" + + name: str + + +class LoadType(Enum): + """Possibly load types""" + + ACTIVE = auto() + REACTIVE = auto() + + @classmethod + def from_str(cls, load_type: Union[str, LoadType, None]) -> Union[LoadType, None]: + """Create loadtype from str""" + if isinstance(load_type, str): + return cls[load_type.upper()] + return load_type + + @classmethod + def check( + cls, load_type: Union[str, LoadType, None] + ) -> Union[Type[LoadType], Set[LoadType]]: + """Check for all loadtypes or specific loadtype""" + return ( + cls + if load_type is None + else {cls[load_type.upper()] if isinstance(load_type, str) else load_type} + ) + + +class LoadModel(Enum): + """ + Members define possible models used. + + Maps model names to 'interp1d' interpolation behaviour. + """ + + RAMP = "linear" + STEP = "previous" + + +@dataclass +class Efficiency: + """Efficiency data container""" + + value: Union[float, Dict[str, float], Dict[LoadType, float]] + description: str = "" + + def __post_init__(self): + """Enforce value structure""" + self.value = efficiency_type(self.value) + + +class EfficiencyDictType(TypedDict): + """Typing for efficiency object""" + + value: Dict[Union[LoadType, str], float] + description: NotRequired[str] + + +class Descriptor: + """Data class property descriptor + + See https://docs.python.org/3/library/dataclasses.html#descriptor-typed-fields + """ + + def __set_name__(self, _, name: str): + """Set the attribute name from a dataclass""" + self._name = f"_{name}" + + +class ActiveReactiveDescriptor(Descriptor): + """Descriptor for setting up active and reactive data dictionaries""" + + def __get__( + self, obj: Any, _ + ) -> Union[Callable[[], Dict[LoadType, np.ndarray]], Dict[LoadType, np.ndarray]]: + """Get the config""" + if obj is None: + return lambda: { + LoadType.ACTIVE: np.arange(2), + LoadType.REACTIVE: np.arange(2), + } + return getattr(obj, self._name) + + def __set__( + self, + obj: Any, + value: Union[ + Callable[[], Dict[LoadType, np.ndarray]], + Dict[Union[LoadType, str], Union[np.ndarray, list]], + np.ndarray, + ], + ): + """Setup the config""" + if callable(value): + value = value() + + if isinstance(value, (np.ndarray, list)): + value = { + LoadType.ACTIVE: np.asarray(value), + LoadType.REACTIVE: np.asarray(value).copy(), + } + else: + ld_t = {LoadType.ACTIVE, LoadType.REACTIVE} + value = { + k if isinstance(k, LoadType) else LoadType[k.upper()]: np.asarray(v) + for k, v in value.items() + } + if missing_keys := ld_t - value.keys(): + value[missing_keys.pop()] = np.zeros_like( + value[(value.keys() - missing_keys).pop()] + ) + setattr(obj, self._name, value) + + +def efficiency_type( + value: Union[float, Dict[str, float], Dict[LoadType, float]], +) -> Dict[LoadType, float]: + """Convert efficiency value to the correct structure""" + if isinstance(value, (float, int)): + value = {LoadType.ACTIVE: value, LoadType.REACTIVE: value} + else: + ld_t = {LoadType.ACTIVE, LoadType.REACTIVE} + value = { + k if isinstance(k, LoadType) else LoadType[k.upper()]: v + for k, v in value.items() + } + if missing_keys := ld_t - value.keys(): + value[missing_keys.pop()] = np.ones_like( + value[(value.keys() - missing_keys).pop()] + ) + return value + + +class PhaseEfficiencyDescriptor(Descriptor): + """Efficiency descriptor for use with dataclasses""" + + def __get__( + self, obj: Any, _ + ) -> Union[Callable[[], Dict], Dict[str, List[Efficiency]]]: + """Get the config""" + if obj is None: + return dict + return getattr(obj, self._name) + + def __set__( + self, + obj: Any, + value: Union[ + Callable[[], Dict], Dict[str, List[Union[Efficiency, EfficiencyDictType]]] + ], + ): + """Setup the config""" + if callable(value): + value = value() + for k, val in value.items(): + for no, v in enumerate(val): + if not isinstance(v, Efficiency): + v["value"] = efficiency_type(v["value"]) + value[k][no] = Efficiency(**v) + + setattr(obj, self._name, value) + + +class LoadEfficiencyDescriptor(Descriptor): + """Efficiency descriptor for use with dataclasses""" + + def __get__(self, obj: Any, _) -> Union[Callable[[], list], List[Efficiency]]: + """Get the config""" + if obj is None: + return list + return getattr(obj, self._name) + + def __set__( + self, + obj: Any, + value: Union[Callable[[], list], List[Union[EfficiencyDictType, Efficiency]]], + ): + """Setup the config""" + if callable(value): + value = value() + for no, v in enumerate(value): + if not isinstance(v, Efficiency): + v["value"] = efficiency_type(v["value"]) + value[no] = Efficiency(**v) + + setattr(obj, self._name, value) + + +@dataclass +class ScenarioConfig(Config): + """Power cycle scenario config""" + + pulses: dict[str, int] + description: str = "" + + +@dataclass +class PulseConfig(Config): + """Power cycle pulse config""" + + phases: list[str] + description: str = "" + + +@dataclass +class PhaseConfig(Config): + """Power cycle phase config""" + + operation: str + subphases: list[str] + description: str = "" + + +@dataclass +class SubPhaseConfig(Config): + """SubPhase Config""" + + duration: Union[float, str] + loads: list[str] = field(default_factory=list) + efficiencies: PhaseEfficiencyDescriptor = PhaseEfficiencyDescriptor() + unit: str = "s" + description: str = "" + reference: str = "" + + def __post_init__(self): + """Enforce unit conversion""" + if isinstance(self.duration, (float, int)): + self.duration = raw_uc(self.duration, self.unit, "second") + self.unit = "s" + + +@dataclass +class SystemConfig(Config): + """Power cycle system config""" + + subsystems: List[str] + description: str = "" + + +@dataclass +class SubSystemConfig(Config): + """Power cycle sub system config""" + + loads: List[str] + description: str = "" + + +@dataclass +class LoadConfig(Config): + """Power cycle load config""" + + time: ActiveReactiveDescriptor = ActiveReactiveDescriptor() + data: ActiveReactiveDescriptor = ActiveReactiveDescriptor() + efficiencies: LoadEfficiencyDescriptor = LoadEfficiencyDescriptor() + model: Union[LoadModel, str] = LoadModel.RAMP + unit: str = "W" + description: str = "" + normalised: bool = True + consumption: bool = True + + def __post_init__(self): + """Validate load""" + if isinstance(self.model, str): + self.model = LoadModel[self.model.upper()] + for lt in LoadType: + if self.data[lt].size != self.time[lt].size: + raise ValueError( + f"time and data must be the same length for {self.name}: " + f"{self.data[lt]}" + ) + if any(np.diff(self.time[lt]) < 0): + raise ValueError("time must increase") + + self.data[lt] = raw_uc(self.data[lt], self.unit, "W") + self.unit = "W" + + def interpolate( + self, + time: npt.ArrayLike, + end_time: Optional[float] = None, + load_type: Union[str, LoadType] = LoadType.ACTIVE, + ) -> np.ndarray: + """ + Interpolate load for a given time vector + + Notes + ----- + The interpolation type is set by load.model. + Any out-of-bound values are set to zero. + """ + if isinstance(load_type, str): + load_type = LoadType[load_type.upper()] + return interp1d( + self.time[load_type], + self.data[load_type], + kind=self.model.value, + bounds_error=False, # turn-off error for out-of-bound + fill_value=(0, 0), # below-/above-bounds extrapolations + )(time if self.normalised or end_time is None else np.array(time) * end_time) + + +def interpolate_extra(vector: npt.NDArray, n_points: int): + """ + Add points between each point in a vector. + """ + if n_points == 0: + return vector + + return np.concatenate([ + *( + np.linspace(vector[s], vector[s + 1], n_points + 1, endpoint=False) + for s in range(len(vector) - 1) + ), + np.atleast_1d(vector[-1]), + ]) + + +def _normalise_timeseries( + time: npt.ArrayLike, + end_time: Optional[float] = None, +) -> Tuple[np.ndarray, Optional[float]]: + time = np.asarray(time) + if min(time) < 0: + raise NotImplementedError("Negative time not supported") + + if max(time) > 1: + mx_time = max(time) + return time / mx_time, mx_time if end_time is None else end_time + return time, end_time + + +class LoadSet: + """LoadSet of a phase""" + + def __init__( + self, + loads: Dict[str, LoadConfig], + ): + self._loads = loads + + @staticmethod + def _consumption_flag(consumption: Optional[bool] = None) -> Set[bool]: + return {True, False} if consumption is None else {consumption} + + def get_load_data_with_efficiencies( + self, + timeseries: np.ndarray, + load_type: Union[str, LoadType], + unit: Optional[str] = None, + end_time: Optional[float] = None, + *, + consumption: Optional[bool] = None, + ) -> Dict[str, np.ndarray]: + """ + Get load data taking into account efficiencies and consumption + + Parameters + ---------- + timeseries: + time array + load_type: + Type of load + unit: + return unit, defaults to [W] or [var] + end_time: + for unnormalised loads this assures the load is + applied at the right point in time + consumption: + return only consumption loads + """ + data = self.get_explicit_data_consumption( + timeseries, load_type, unit, end_time, consumption=consumption + ) + load_check = LoadType.check(load_type) + for ld_name in data: + load_conf = self._loads[ld_name] + for eff in load_conf.efficiencies: + for eff_type, eff_val in eff.value.items(): + if eff_type in load_check: + data[ld_name] *= eff_val + + return data + + def get_explicit_data_consumption( + self, + timeseries: np.ndarray, + load_type: Union[str, LoadType], + unit: Optional[str] = None, + end_time: Optional[float] = None, + *, + consumption: Optional[bool] = None, + ) -> Dict[str, np.ndarray]: + """ + Get data with consumption resulting in an oppositely signed load + + Parameters + ---------- + timeseries: + time array + load_type: + Type of load + unit: + return unit, defaults to [W] or [var] + end_time: + for unnormalised loads this assures the load is + applied at the right point in time + consumption: + return only consumption loads + """ + loads = self.get_interpolated_loads( + timeseries, load_type, unit, end_time, consumption=consumption + ) + return { + ld_n: -ld if self._loads[ld_n].consumption else ld + for ld_n, ld in loads.items() + } + + def build_timeseries( + self, + load_type: Optional[Union[str, LoadType]] = None, + end_time: Optional[float] = None, + *, + consumption: Optional[bool] = None, + ) -> np.ndarray: + """Build a combined time series based on loads""" + times = [] + for load in self._loads.values(): + for time in self._gettime( + load, load_type, self._consumption_flag(consumption) + ): + if load.normalised: + times.append(time) + else: + times.append(time / (max(time) if end_time is None else end_time)) + return np.unique(np.concatenate(times)) + + @staticmethod + def _gettime( + load, load_type: Optional[Union[str, LoadType]], consumption_check: Set[bool] + ): + load_type = LoadType.from_str(load_type) + if load.consumption in consumption_check: + if load_type is None: + yield from load.time.values() + else: + yield load.time[load_type] + + def get_interpolated_loads( + self, + timeseries: np.ndarray, + load_type: Union[str, LoadType], + unit: Optional[str] = None, + end_time: Optional[float] = None, + *, + consumption: Optional[bool] = None, + ) -> Dict[str, np.ndarray]: + """ + Get loads for a given time series + + Parameters + ---------- + timeseries: + time array + load_type: + Type of load + unit: + return unit, defaults to [W] or [var] + end_time: + for unnormalised loads this assures the load is + applied at the right point in time + consumption: + return only consumption loads + """ + timeseries, end_time = _normalise_timeseries(timeseries, end_time) + load_type = LoadType.from_str(load_type) + _cnsmptn = self._consumption_flag(consumption) + + return { + load.name: load.interpolate(timeseries, end_time, load_type) + if unit is None + else raw_uc( + load.interpolate(timeseries, end_time, load_type), + load.unit, + unit, + ) + for load in self._loads.values() + if load.consumption in _cnsmptn + } + + def load_total( + self, + timeseries: np.ndarray, + load_type: Union[str, LoadType], + unit: Optional[str] = None, + end_time: Optional[float] = None, + *, + consumption: Optional[bool] = None, + ) -> np.ndarray: + """Total load for each timeseries point for a given load_type + + Parameters + ---------- + timeseries: + time array + load_type: + Type of load + unit: + return unit, defaults to [W] or [var] + end_time: + for unnormalised loads this assures the load is + applied at the right point in time + consumption: + return only consumption loads + """ + return np.sum( + list( + self.get_load_data_with_efficiencies( + timeseries, load_type, unit, end_time, consumption=consumption + ).values() + ), + axis=0, + ) + + +@dataclass +class Phase: + """Phase container""" + + config: PhaseConfig + subphases: Dict[str, SubPhaseConfig] + loads: LoadSet + + def __post_init__(self): + """Validate duration""" + if self.duration < 0: + raise ValueError( + f"{self.config.name} phase duration must be positive: {self.duration}s" + ) + + @property + def duration(self): + """Duration of phase""" + return getattr(np, self.config.operation)([ + s_ph.duration for s_ph in self.subphases.values() + ]) + + def _process_phase_efficiencies( + self, loads: Dict[str, np.ndarray], load_type: Union[str, LoadType] + ): + load_check = LoadType.check(load_type) + for subphase in self.subphases.values(): + self._find_duplicate_loads(subphase, loads) + for eff_name, effs in subphase.efficiencies.items(): + for eff in effs: + for eff_type, eff_val in eff.value.items(): + if eff_type in load_check and eff_name in loads: + loads[eff_name] *= eff_val + return loads + + @staticmethod + def _find_duplicate_loads(subphase: SubPhaseConfig, loads: Dict[str, np.ndarray]): + """Add duplication efficiency. + + If a load is duplicated in subphase.loads, + the resultant data array is multiplied by the number of repeats + """ + u, c = np.unique(subphase.loads, return_counts=True) + counts = c[c > 1] + for cnt, dup in enumerate(u[c > 1]): + if dup in loads: + eff = counts[cnt] + bluemira_debug(f"Duplicate load {dup}, duplication efficiency of {eff}") + loads[dup] *= eff + + def build_timeseries( + self, + load_type: Optional[Union[str, LoadType]] = None, + *, + consumption: Optional[bool] = None, + ) -> np.ndarray: + """Build a combined time series based on loads""" + return ( + self.loads.build_timeseries( + load_type=load_type, end_time=self.duration, consumption=consumption + ) + * self.duration + ) + + def load_total( + self, + timeseries: np.ndarray, + load_type: Union[str, LoadType], + unit: Optional[str] = None, + *, + consumption: Optional[bool] = None, + ) -> np.ndarray: + """Total load for each timeseries point for a given load_type + + Parameters + ---------- + timeseries: + time array + load_type: + Type of load + unit: + return unit, defaults to [W] or [var] + consumption: + return only consumption loads + """ + timeseries, _ = _normalise_timeseries(timeseries, self.duration) + return np.sum( + list( + self.get_load_data_with_efficiencies( + timeseries, load_type, unit, consumption=consumption + ).values() + ), + axis=0, + ) + + def get_load_data_with_efficiencies( + self, + timeseries: np.ndarray, + load_type: Union[str, LoadType], + unit: Optional[str] = None, + *, + consumption: Optional[bool] = None, + ) -> Dict[str, np.ndarray]: + """ + Get load data taking into account efficiencies and consumption + + Parameters + ---------- + timeseries: + time array + load_type: + Type of load + unit: + return unit, defaults to [W] or [var] + consumption: + return only consumption loads + """ + timeseries, _ = _normalise_timeseries(timeseries, self.duration) + return self._process_phase_efficiencies( + self.loads.get_load_data_with_efficiencies( + timeseries, + load_type, + unit, + end_time=self.duration, + consumption=consumption, + ), + load_type, + ) + + +class PulseDictType(TypedDict): + """Pulse dictionary typing""" + + repeat: int + data: Dict[str, Phase] + + +class LibraryConfig: + """Power Cycle Configuration""" + + def __init__( + self, + scenario: ScenarioConfig, + pulse: Dict[str, PulseConfig], + phase: Dict[str, PhaseConfig], + subphase: Dict[str, SubPhaseConfig], + system: Dict[str, SystemConfig], + subsystem: Dict[str, SubSystemConfig], + loads: Dict[str, LoadConfig], + durations: Optional[Dict[str, float]] = None, + ): + self.scenario = scenario + self.pulse = pulse + self.phase = phase + self.subphase = subphase + self.system = system + self.subsystem = subsystem + self.loads = loads + self._import_subphase_duration(durations) + + def check_config(self): + """Check powercycle configuration""" + ph_keys = self.phase.keys() + sph_keys = self.subphase.keys() + ss_keys = self.subsystem.keys() + loads = self.loads.keys() + # scenario has known pulses + if unknown_pulse := self.scenario.pulses.keys() - self.pulse.keys(): + raise ValueError(f"Unknown pulses {unknown_pulse}") + # pulses have known phases + for pulse in self.pulse.values(): + if unknown_phase := pulse.phases - ph_keys: + raise ValueError(f"Unknown phases {unknown_phase}") + # phases have known subphases + for ph_c in self.phase.values(): + if unknown_s_ph := ph_c.subphases - sph_keys: + raise ValueError(f"Unknown subphase configurations {unknown_s_ph}") + # subphases have known loads + for subphase in self.subphase.values(): + if unknown_load := subphase.loads - loads: + raise ValueError(f"Unknown loads {unknown_load}") + + # systems have known subsystems + for sys_c in self.system.values(): + if unknown := sys_c.subsystems - ss_keys: + raise ValueError(f"Unknown subsystem configurations {unknown}") + # subsystems have known loads + for s_sys_c in self.subsystem.values(): + if unknown_load := s_sys_c.loads - loads: + raise ValueError(f"Unknown loads {unknown_load}") + + def _import_subphase_duration( + self, subphase_duration_params: Optional[Dict[str, float]] = None + ): + """Import subphase data""" + for s_ph in self.subphase.values(): + if isinstance(s_ph.duration, str): + key = s_ph.duration.replace("$", "") + if subphase_duration_params is None: + raise KeyError(key) + s_ph.duration = subphase_duration_params[key] + + def add_load_config( + self, + load: LoadConfig, + subphases: Optional[Union[str, Iterable[str]]] = None, + subphase_efficiency: Optional[List[Efficiency]] = None, + ): + """Add load config""" + self.loads[load.name] = load + self.link_load_to_subphase(load.name, subphases or [], subphase_efficiency) + + def link_load_to_subphase( + self, + load_name: str, + subphases: Union[str, Iterable[str]], + subphase_efficiency: Optional[List[Efficiency]] = None, + ): + """Link a load to a specific subphase""" + if isinstance(subphases, str): + subphases = [subphases] + for subphase in subphases: + self.subphase[subphase].loads.append(load_name) + if subphase_efficiency is not None: + self.subphase[subphase].efficiencies[load_name] = subphase_efficiency + + @classmethod + def from_json( + cls, + manager_config_path: Union[Path, str], + durations: Optional[Dict[str, float]] = None, + ): + """Create configuration from pure json""" + return cls.from_dict(read_json(manager_config_path), durations) + + @classmethod + def from_dict( + cls, data: Dict[str, Any], durations: Optional[Dict[str, float]] = None + ): + """Create configuration from dictionary""" + return cls( + scenario=ScenarioConfig(**data["scenario"]), + pulse={ + k: PulseConfig(name=k, **v) for k, v in data["pulse_library"].items() + }, + phase={ + k: PhaseConfig(name=k, **v) for k, v in data["phase_library"].items() + }, + subphase={ + k: SubPhaseConfig(name=k, **v) + for k, v in data["subphase_library"].items() + }, + system={ + k: SystemConfig(name=k, **v) for k, v in data["system_library"].items() + }, + subsystem={ + k: SubSystemConfig(name=k, **v) + for k, v in data["sub_system_library"].items() + }, + loads={k: LoadConfig(name=k, **v) for k, v in data["load_library"].items()}, + durations=durations, + ) + + def get_phase(self, phase: str, *, check=True) -> Phase: + """Create a single phase object""" + if check: + self.check_config() + + phase_config = self.phase[phase] + subphases = {k: self.subphase[k] for k in phase_config.subphases} + + return Phase( + phase_config, + subphases, + LoadSet({ + ld: self.loads[ld] + for subphase in subphases.values() + for ld in subphase.loads + }), + ) + + def get_pulse(self, pulse: str, *, check=True) -> Dict[str, Phase]: + """Create a pulse dictionary""" + if check: + self.check_config() + return { + phase: self.get_phase(phase, check=False) + for phase in self.pulse[pulse].phases + } + + def get_scenario(self) -> Dict[str, PulseDictType]: + """Create a scenario dictionary""" + self.check_config() + return { + pulse: {"repeat": reps, "data": self.get_pulse(pulse, check=False)} + for pulse, reps in self.scenario.pulses.items() + } diff --git a/bluemira/power_cycle/tools.py b/bluemira/power_cycle/tools.py new file mode 100644 index 0000000000..4cfae91cc1 --- /dev/null +++ b/bluemira/power_cycle/tools.py @@ -0,0 +1,33 @@ +# SPDX-FileCopyrightText: 2021-present M. Coleman, J. Cook, F. Franza +# SPDX-FileCopyrightText: 2021-present I.A. Maione, S. McIntosh +# SPDX-FileCopyrightText: 2021-present J. Morris, D. Short +# +# SPDX-License-Identifier: LGPL-2.1-or-later + +""" +Utility functions for the power cycle model. +""" + +import json +from typing import Any, Dict + +import matplotlib.pyplot as plt + + +def read_json(file_path) -> Dict[str, Any]: + """ + Returns the contents of a 'json' file. + """ + with open(file_path) as json_file: + return json.load(json_file) + + +def create_axes(ax=None): + """ + Create axes object. + + If 'None', creates a new 'axes' instance. + """ + if ax is None: + _, ax = plt.subplots() + return ax diff --git a/documentation/source/examples.rst b/documentation/source/examples.rst index a5f8887435..3a8201542c 100644 --- a/documentation/source/examples.rst +++ b/documentation/source/examples.rst @@ -75,6 +75,7 @@ Other Examples examples/base/units examples/balance_of_plant/steady_state_example + examples/power_cycle/simple_example examples/fuel_cycle/EUDEMO_fuelcycle examples/materials/material_definitions examples/mesh/mesh_tutorial diff --git a/examples/power_cycle/scenario_config.json b/examples/power_cycle/scenario_config.json new file mode 100644 index 0000000000..234e066399 --- /dev/null +++ b/examples/power_cycle/scenario_config.json @@ -0,0 +1,399 @@ +{ + "Power Cycle": { + "scenario": { + "name": "std_scenario", + "description": "Standard scenario", + "pulses": { "std": 1 } + }, + "pulse_library": { + "std": { + "description": "Standard pulse", + "phases": ["dwl", "d2f", "ftt", "f2d"] + } + }, + "phase_library": { + "dwl": { + "description": "Dwell", + "operation": "max", + "subphases": ["csr", "pmp"] + }, + "d2f": { + "description": "Transition between dwell and flat-top", + "operation": "sum", + "subphases": ["cru", "bri"] + }, + "ftt": { + "description": "Flat-top", + "operation": "sum", + "subphases": ["plb"] + }, + "f2d": { + "description": "Transition between flat-top and dwell", + "operation": "sum", + "subphases": ["brt", "crd"] + } + }, + "subphase_library": { + "csr": { + "description": "Central Solenoid recharge", + "duration": "$cs_recharge_time", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": { "reactive": 5, "active": 5 }, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + } + }, + "pmp": { + "description": "Reactor pump-down", + "duration": "$pumpdown_time", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + } + }, + "cru": { + "description": "Current ramp-up", + "duration": "$ramp_up_time", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + } + }, + "crd": { + "description": "Current ramp-down", + "duration": "$ramp_down_time", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ], + "hcpb_hirn": [ + { "value": { "reactive": 0 }, "description": "no reactive load" } + ] + } + }, + "bri": { + "description": "Burn initiation (heat up plasma)", + "duration": 19, + "unit": "second", + "loads": [ + "vv", + "hcpb_vvps", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + }, + "reference": "" + }, + "brt": { + "description": "Burn termination (cool down plasma)", + "duration": 123, + "unit": "second", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ], + "hcpb_hirn": [ + { "value": { "reactive": 0 }, "description": "no reactive load" } + ] + }, + "reference": "" + }, + "plb": { + "description": "Plasma burn", + "duration": 2, + "unit": "hour", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + }, + "reference": "" + } + }, + "system_library": { + "BOP": { + "description": "Magnetics (PBS = 49,50,54,58,70)", + "subsystems": ["PHT", "IHT", "PCS"] + }, + "AUX": { + "description": "Auxiliaries (PBS = 80,82,83,87)", + "subsystems": ["EPS", "BSU", "OTR"] + } + }, + "sub_system_library": { + "PHT": { + "description": "Primary Heat Transfer System", + "loads": ["vv", "hcpb_vvps", "div_lim", "hcpb_bb_peak"] + }, + "IHT": { + "description": "Intermediate Heat Transfer System", + "loads": [] + }, + "PCS": { + "description": "Power Conversion System", + "loads": ["hcpb_hirn", "turb"] + }, + "EPS": { + "description": "Electrical Power Supply", + "loads": ["eps_peak", "eps_upk"] + }, + "BSU": { + "description": "Buildings & Site Utilities", + "loads": ["buildings", "site"] + }, + "OTR": { + "description": "Other Auxiliary Systems", + "loads": ["sf6_air"] + } + }, + "load_library": { + "vv": { + "time": [0, 1], + "data": { "reactive": [4.7, 4.7], "active": [9.7, 9.7] }, + "unit": "MW", + "model": "RAMP", + "normalised": true, + "consumption": true, + "efficiencies": [], + "description": "Vacuum Vessel (VV)" + }, + "hcpb_vvps": { + "time": { "active": [0, 1] }, + "data": { "active": [2.3, 2.3] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true, + "description": "(HCPB) Vacuum Vessel pressure suppression (VVPS)" + }, + "div_lim": { + "time": [0, 1], + "data": { "reactive": [12.1, 12.1], "active": [19.5, 19.5] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true, + "description": "Divertor & Limiter (DV & LM)" + }, + "hcpb_bb_peak": { + "description": "(HCPB) Breeding Blanket (BB) - peak value", + "time": [0, 1], + "data": { "reactive": [54.4, 54.4], "active": [165.6, 165.6] }, + "unit": "MW", + "model": "RAMP", + "normalised": true + }, + "hcpb_hirn": { + "description": "(HCPB) Hirn Cycle components", + "time": [0, 1], + "data": { "reactive": [5.8, 5.8], "active": [12, 12] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true + }, + "turb": { + "description": "nominal turbine power for direct concept", + "time": [0, 1], + "data": [750, 750], + "unit": "MW", + "model": "RAMP", + "efficiencies": [{ "value": 0.85, "description": "indirect storage" }], + "consumption": false, + "normalised": true + }, + "eps_peak": { + "description": "peak value extrapolated to all relevant phases", + "time": [0, 1], + "data": [300, 300], + "unit": "MW", + "model": "STEP" + }, + "eps_upk": { + "description": "EPS upkeep", + "time": [0, 120], + "data": { "reactive": [10.2, 10.2], "active": [21, 21] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": false + }, + "buildings": { + "description": "buildings (light/elevators/HVAC; ITER extrapolation)", + "time": [0, 1], + "data": { "reactive": [26.6, 26.6], "active": [54.8, 54.8] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true + }, + "site": { + "description": "site utilities", + "time": [0, 1], + "data": { "reactive": [1.9, 1.9], "active": [3.1, 3.1] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true + }, + "sf6_air": { + "description": "ITER extrapolation for: SF6 & compressed air distribution, CCWS, CHWS", + "time": [0, 1], + "data": { "reactive": [56.4, 56.4], "active": [90.9, 90.9] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true + } + } + } +} diff --git a/examples/power_cycle/simple_example.ex.py b/examples/power_cycle/simple_example.ex.py new file mode 100644 index 0000000000..c6aecd2775 --- /dev/null +++ b/examples/power_cycle/simple_example.ex.py @@ -0,0 +1,90 @@ +# --- +# jupyter: +# jupytext: +# cell_metadata_filter: tags,-all +# notebook_metadata_filter: -jupytext.text_representation.jupytext_version +# text_representation: +# extension: .py +# format_name: percent +# format_version: '1.3' +# kernelspec: +# display_name: Python 3 (ipykernel) +# language: python +# name: python3 +# --- + +# %% tags=["remove-cell"] +# SPDX-FileCopyrightText: 2021-present M. Coleman, J. Cook, F. Franza +# SPDX-FileCopyrightText: 2021-present I.A. Maione, S. McIntosh +# SPDX-FileCopyrightText: 2021-present J. Morris, D. Short +# +# SPDX-License-Identifier: LGPL-2.1-or-later +"""Power cycle example""" +# %% + +from __future__ import annotations + +from pathlib import Path + +from bluemira.base.reactor_config import ReactorConfig +from bluemira.power_cycle.net import ( + Efficiency, + LibraryConfig, + LoadConfig, + interpolate_extra, +) + +# %% [markdown] +# # Power Cycle example +# +# Firstly we read in the build config and extract the config for the PowerCycle. +# We import any subphase durations needed for the config. +# In principle these could come from other parts of the reactor design. +# %% + +reactor_config = ReactorConfig(Path(__file__).parent / "scenario_config.json", None) +config = LibraryConfig.from_dict( + reactor_config.config_for("Power Cycle"), + durations={ + "cs_recharge_time": 300, + "pumpdown_time": 600, + "ramp_up_time": 157, + "ramp_down_time": 157, + }, +) + +# %% [markdown] +# We can then dynamically add a new load to a specific subphase of the config. +# %% + +config.add_load_config( + load=LoadConfig( + "cs_power", + data={"active": [1, 2], "reactive": [10, 20]}, + efficiencies=[Efficiency(0.1)], + description="something made up", + ), + subphases=["cru", "bri"], + subphase_efficiency=[Efficiency({"reactive": 0.2})], +) + +# %% [markdown] +# Once the config is created we can now pull out the data for a specific phase. +# +# Below we have interpolated the timeseries and pulled out the active and reactive loads +# %% + +phase = config.get_phase("dwl") + +timeseries = interpolate_extra(phase.build_timeseries(), 1) + +active_loads = phase.get_load_data_with_efficiencies(timeseries, "active", "MW") +active_load_total = phase.load_total(timeseries, "active", "MW") + +# %% [markdown] +# Note for reactive loads the unit is 'var' (volt-ampere reactive). Although numerically +# identical to a watt it is the wrong unit for reactive loads. +# %% + +reactive_loads = phase.get_load_data_with_efficiencies(timeseries, "reactive", "Mvar") +reactive_load_total = phase.load_total(timeseries, "reactive", "Mvar") diff --git a/tests/power_cycle/test_data/scenario_config.json b/tests/power_cycle/test_data/scenario_config.json new file mode 100644 index 0000000000..234e066399 --- /dev/null +++ b/tests/power_cycle/test_data/scenario_config.json @@ -0,0 +1,399 @@ +{ + "Power Cycle": { + "scenario": { + "name": "std_scenario", + "description": "Standard scenario", + "pulses": { "std": 1 } + }, + "pulse_library": { + "std": { + "description": "Standard pulse", + "phases": ["dwl", "d2f", "ftt", "f2d"] + } + }, + "phase_library": { + "dwl": { + "description": "Dwell", + "operation": "max", + "subphases": ["csr", "pmp"] + }, + "d2f": { + "description": "Transition between dwell and flat-top", + "operation": "sum", + "subphases": ["cru", "bri"] + }, + "ftt": { + "description": "Flat-top", + "operation": "sum", + "subphases": ["plb"] + }, + "f2d": { + "description": "Transition between flat-top and dwell", + "operation": "sum", + "subphases": ["brt", "crd"] + } + }, + "subphase_library": { + "csr": { + "description": "Central Solenoid recharge", + "duration": "$cs_recharge_time", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": { "reactive": 5, "active": 5 }, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + } + }, + "pmp": { + "description": "Reactor pump-down", + "duration": "$pumpdown_time", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + } + }, + "cru": { + "description": "Current ramp-up", + "duration": "$ramp_up_time", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + } + }, + "crd": { + "description": "Current ramp-down", + "duration": "$ramp_down_time", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ], + "hcpb_hirn": [ + { "value": { "reactive": 0 }, "description": "no reactive load" } + ] + } + }, + "bri": { + "description": "Burn initiation (heat up plasma)", + "duration": 19, + "unit": "second", + "loads": [ + "vv", + "hcpb_vvps", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + }, + "reference": "" + }, + "brt": { + "description": "Burn termination (cool down plasma)", + "duration": 123, + "unit": "second", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "hcpb_bb_peak": [ + { + "value": 5, + "description": "coolant flow to 20%" + } + ], + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ], + "hcpb_hirn": [ + { "value": { "reactive": 0 }, "description": "no reactive load" } + ] + }, + "reference": "" + }, + "plb": { + "description": "Plasma burn", + "duration": 2, + "unit": "hour", + "loads": [ + "vv", + "hcpb_vvps", + "div_lim", + "hcpb_bb_peak", + "hcpb_hirn", + "turb", + "eps_peak", + "eps_upk", + "buildings", + "site", + "sf6_air" + ], + "efficiencies": { + "turb": [ + { + "value": { "reactive": 1.25 }, + "description": "inverse of minimum power factor @ 0.8" + } + ] + }, + "reference": "" + } + }, + "system_library": { + "BOP": { + "description": "Magnetics (PBS = 49,50,54,58,70)", + "subsystems": ["PHT", "IHT", "PCS"] + }, + "AUX": { + "description": "Auxiliaries (PBS = 80,82,83,87)", + "subsystems": ["EPS", "BSU", "OTR"] + } + }, + "sub_system_library": { + "PHT": { + "description": "Primary Heat Transfer System", + "loads": ["vv", "hcpb_vvps", "div_lim", "hcpb_bb_peak"] + }, + "IHT": { + "description": "Intermediate Heat Transfer System", + "loads": [] + }, + "PCS": { + "description": "Power Conversion System", + "loads": ["hcpb_hirn", "turb"] + }, + "EPS": { + "description": "Electrical Power Supply", + "loads": ["eps_peak", "eps_upk"] + }, + "BSU": { + "description": "Buildings & Site Utilities", + "loads": ["buildings", "site"] + }, + "OTR": { + "description": "Other Auxiliary Systems", + "loads": ["sf6_air"] + } + }, + "load_library": { + "vv": { + "time": [0, 1], + "data": { "reactive": [4.7, 4.7], "active": [9.7, 9.7] }, + "unit": "MW", + "model": "RAMP", + "normalised": true, + "consumption": true, + "efficiencies": [], + "description": "Vacuum Vessel (VV)" + }, + "hcpb_vvps": { + "time": { "active": [0, 1] }, + "data": { "active": [2.3, 2.3] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true, + "description": "(HCPB) Vacuum Vessel pressure suppression (VVPS)" + }, + "div_lim": { + "time": [0, 1], + "data": { "reactive": [12.1, 12.1], "active": [19.5, 19.5] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true, + "description": "Divertor & Limiter (DV & LM)" + }, + "hcpb_bb_peak": { + "description": "(HCPB) Breeding Blanket (BB) - peak value", + "time": [0, 1], + "data": { "reactive": [54.4, 54.4], "active": [165.6, 165.6] }, + "unit": "MW", + "model": "RAMP", + "normalised": true + }, + "hcpb_hirn": { + "description": "(HCPB) Hirn Cycle components", + "time": [0, 1], + "data": { "reactive": [5.8, 5.8], "active": [12, 12] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true + }, + "turb": { + "description": "nominal turbine power for direct concept", + "time": [0, 1], + "data": [750, 750], + "unit": "MW", + "model": "RAMP", + "efficiencies": [{ "value": 0.85, "description": "indirect storage" }], + "consumption": false, + "normalised": true + }, + "eps_peak": { + "description": "peak value extrapolated to all relevant phases", + "time": [0, 1], + "data": [300, 300], + "unit": "MW", + "model": "STEP" + }, + "eps_upk": { + "description": "EPS upkeep", + "time": [0, 120], + "data": { "reactive": [10.2, 10.2], "active": [21, 21] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": false + }, + "buildings": { + "description": "buildings (light/elevators/HVAC; ITER extrapolation)", + "time": [0, 1], + "data": { "reactive": [26.6, 26.6], "active": [54.8, 54.8] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true + }, + "site": { + "description": "site utilities", + "time": [0, 1], + "data": { "reactive": [1.9, 1.9], "active": [3.1, 3.1] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true + }, + "sf6_air": { + "description": "ITER extrapolation for: SF6 & compressed air distribution, CCWS, CHWS", + "time": [0, 1], + "data": { "reactive": [56.4, 56.4], "active": [90.9, 90.9] }, + "unit": "MW", + "model": "RAMP", + "consumption": true, + "normalised": true + } + } + } +} diff --git a/tests/power_cycle/test_net.py b/tests/power_cycle/test_net.py new file mode 100644 index 0000000000..e4d3cd8577 --- /dev/null +++ b/tests/power_cycle/test_net.py @@ -0,0 +1,279 @@ +# SPDX-FileCopyrightText: 2021-present M. Coleman, J. Cook, F. Franza +# SPDX-FileCopyrightText: 2021-present I.A. Maione, S. McIntosh +# SPDX-FileCopyrightText: 2021-present J. Morris, D. Short +# +# SPDX-License-Identifier: LGPL-2.1-or-later +from copy import deepcopy +from pathlib import Path + +import numpy as np +import pytest + +from bluemira.base.reactor_config import ReactorConfig +from bluemira.power_cycle.net import ( + Efficiency, + LibraryConfig, + LoadConfig, + LoadSet, + LoadType, + Phase, + PhaseConfig, + SubPhaseConfig, + interpolate_extra, +) + + +def test_LoadType_from_str(): + assert LoadType.from_str("active") == LoadType.ACTIVE + assert LoadType.from_str("reactive") == LoadType.REACTIVE + + +def test_interpolate_extra_returns_the_correct_length(): + arr = interpolate_extra(np.arange(5), 5) + assert arr.size == 25 + assert max(arr) == 4 + assert min(arr) == 0 + + +def test_SubPhaseConfig_duration(): + pcb = SubPhaseConfig("name", 5, unit="hours") + assert pcb.duration == 18000 + assert pcb.unit == "s" + + +class TestSubLoad: + def test_interpolate(self): + pcsl = LoadConfig( + "name", np.array([0, 0.5, 1]), {"reactive": np.arange(3)}, model="ramp" + ) + assert np.allclose( + pcsl.interpolate([0, 0.1, 0.2, 0.3, 1], load_type="reactive"), + np.array([0, 0.2, 0.4, 0.6, 2]), + ) + pcsl2 = LoadConfig("name", np.array([0, 0.5, 1]), np.arange(3), model="ramp") + assert np.allclose( + pcsl2.interpolate([0, 0.1, 0.2, 0.3, 1], load_type="reactive"), + np.array([0, 0.2, 0.4, 0.6, 2]), + ) + assert np.allclose( + pcsl2.interpolate([0, 0.1, 0.2, 0.3, 1], load_type="active"), + np.array([0, 0.2, 0.4, 0.6, 2]), + ) + + pcsl = LoadConfig( + "name", + np.array([0, 0.5, 2]), + data={"active": np.arange(3)}, + model="ramp", + normalised=False, + ) + assert np.allclose( + pcsl.interpolate([0, 0.1, 0.2, 0.3, 1], 10), + np.array([0, 1 + 1 / 3, 2, 0, 0]), + ) + + def test_validation_raises_ValueError(self): + with pytest.raises(ValueError, match="time and data"): + LoadConfig("name", np.array([0, 0.1, 1]), np.zeros(2), model="ramp") + + with pytest.raises(ValueError, match="time and data"): + LoadConfig("name", [0, 0.1, 1], np.zeros(2), model="ramp") + + with pytest.raises(ValueError, match="time must increase"): + LoadConfig("name", [0, 1, 0.1], np.zeros(3), model="ramp") + + pcsl = LoadConfig("name", [0, 0.1, 1], np.zeros(3), model="ramp", unit="MW") + assert np.allclose(pcsl.time[LoadType.ACTIVE], np.array([0, 0.1, 1])) + assert np.allclose(pcsl.time[LoadType.REACTIVE], np.array([0, 0.1, 1])) + assert pcsl.unit == "W" + + +class TestLoadSet: + @classmethod + def setup_class(cls): + reactor_config = ReactorConfig( + Path(__file__).parent / "test_data" / "scenario_config.json", None + ) + cls._config = LibraryConfig.from_dict( + reactor_config.config_for("Power Cycle"), + { + "cs_recharge_time": 300, + "pumpdown_time": 600, + "ramp_up_time": 157, + "ramp_down_time": 157, + }, + ) + cls._loads = cls._config.get_phase("dwl").loads + + def setup_method(self): + self.loads = deepcopy(self._loads) + + @pytest.mark.parametrize("load_type", ["active", "reactive", None]) + @pytest.mark.parametrize("end_time", [200, None]) + @pytest.mark.parametrize("consumption", [True, False, None]) + def test_build_timeseries(self, load_type, end_time, consumption): + assert np.allclose( + self.loads.build_timeseries( + load_type=load_type, end_time=end_time, consumption=consumption + ), + [0, 0.6, 1] if consumption in {True, None} and end_time == 200 else [0, 1], + ) + + @pytest.mark.parametrize( + ("time", "et", "res1", "res2"), + [ + ( + np.array([0, 0.005, 0.6, 1]), + 200, + np.full(4, -4.7), + np.array([-10.2, -10.2, -10.2, -0.0]), + ), + ( + np.array([0, 0.005, 0.6, 1]), + None, + np.full(4, -4.7), + np.full(4, -10.2), + ), + ( + np.array([0, 1, 120, 200]), + None, + np.full(4, -4.7), + np.array([-10.2, -10.2, -10.2, -0.0]), + ), + ], + ) + def test_get_load_data_with_efficiencies(self, time, et, res1, res2): + load_data = self.loads.get_load_data_with_efficiencies( + time, "reactive", "MW", end_time=et + ) + assert np.allclose(load_data["vv"], res1) + # not normalised + assert np.allclose(load_data["eps_upk"], res2) + + @pytest.mark.parametrize( + ("time", "et", "res"), + [ + (np.array([0, 0.6, 1]), 200, np.array([165.4, 165.4, 175.6])), + (np.array([0, 120, 200]), None, np.array([165.4, 165.4, 175.6])), + (np.array([0, 0.6, 1]), None, np.full(3, 165.4)), + ], + ) + def test_get_load_total(self, time, et, res): + assert np.allclose( + self.loads.load_total(time, "reactive", "MW", end_time=et), res + ) + + +class TestPhase: + def setup_method(self): + self.phase = Phase( + PhaseConfig("dwl", "max", ["a", "b"]), + { # the loads list is not checked again...should we? + "a": SubPhaseConfig("a", 5, ["name", "name"]), + "b": SubPhaseConfig("b", 10, ["name2"], {"name2": [Efficiency(0.1)]}), + }, + LoadSet({ + "name": LoadConfig( + "name", np.array([0, 0.5, 1]), np.arange(3), model="ramp" + ), + "name2": LoadConfig( + "name2", + np.array([0, 0.2, 1]), + np.arange(3), + model="ramp", + consumption=False, + ), + }), + ) + + def test_duration_validation_and_extraction(self): + assert self.phase.duration == 10 + + @pytest.mark.parametrize("load_type", ["active", "reactive", None]) + @pytest.mark.parametrize("consumption", [True, False, None]) + def test_build_timeseries(self, load_type, consumption): + assert np.allclose( + self.phase.build_timeseries(load_type=load_type, consumption=consumption), + [0, 2, 5, 10] + if consumption is None + else [0, 5, 10] + if consumption + else [0, 2, 10], + ) + + @pytest.mark.parametrize("consumption", [True, False, None]) + @pytest.mark.parametrize("load_type", ["active", "reactive"]) + def test_load_total(self, load_type, consumption): + assert np.allclose( + self.phase.load_total([0, 2, 5, 10], load_type, consumption=consumption), + [0.0, -0.7, -1.8625, -3.8] + if consumption is None + else [0.0, -0.8, -2.0, -4.0] + if consumption + else [0.0, 0.1, 0.1375, 0.2], + ) + + @pytest.mark.parametrize("consumption", [True, False, None]) + @pytest.mark.parametrize("load_type", ["active", "reactive"]) + def test_get_load_data_with_efficiencies(self, load_type, consumption): + for _ in range(2): # run the duplicates twice doesnt keep doubling load + res = self.phase.get_load_data_with_efficiencies( + [0, 2, 5, 10], load_type, consumption=consumption + ) + if (name := res.get("name", None)) is not None: + assert np.allclose(name, np.array([-0.0, -0.8, -2.0, -4.0])) + if (name2 := res.get("name2", None)) is not None: + assert np.allclose(name2, np.array([0.0, 0.1, 0.1375, 0.2])) + + +class TestLibraryConfig: + @classmethod + def setup_class(cls): + reactor_config = ReactorConfig( + Path(__file__).parent / "test_data" / "scenario_config.json", None + ) + cls._config = LibraryConfig.from_dict( + reactor_config.config_for("Power Cycle"), + { + "cs_recharge_time": 300, + "pumpdown_time": 600, + "ramp_up_time": 157, + "ramp_down_time": 157, + }, + ) + + def setup_method(self): + self.config = deepcopy(self._config) + + def test_get_scenario(self): + scenario = self.config.get_scenario() + + assert scenario["std"]["repeat"] == 1 + assert len(scenario["std"]["data"].keys()) == 4 + assert all(isinstance(val, Phase) for val in scenario["std"]["data"].values()) + assert scenario["std"]["data"]["dwl"].subphases.keys() == {"csr", "pmp"} + + sph = scenario["std"]["data"]["dwl"].subphases + assert scenario["std"]["data"]["dwl"].loads._loads.keys() == set( + sph["csr"].loads + sph["pmp"].loads + ) + + def test_import_subphase_data(self): + assert self.config.subphase["csr"].duration == 300 + + def test_add_load_config(self): + self.config.add_load_config( + LoadConfig( + "cs_power", + [0, 1], + [10, 20], + model="RAMP", + unit="MW", + description="dunno", + ), + ["cru", "bri"], + ) + assert np.allclose( + self.config.loads["cs_power"].data[LoadType.REACTIVE], [10e6, 20e6] + ) + assert self.config.loads["cs_power"].unit == "W"