From 720afad2fc84fee2c62f675bb7d10c3967bdce44 Mon Sep 17 00:00:00 2001 From: Frode Helgetun Krogh <70878501+frodehk@users.noreply.github.com> Date: Fri, 13 Dec 2024 13:22:02 +0100 Subject: [PATCH 1/2] refactor: move to energy components --- .../domain/infrastructure/__init__.py | 12 +- .../domain/infrastructure/components.py | 1012 ----------------- .../energy_components/asset/__init__.py | 0 .../energy_components/asset/asset.py | 50 + .../energy_components/base/component_dto.py | 122 ++ .../energy_components/common.py | 66 ++ .../compressor/component_dto.py | 31 + .../consumer_system/__init__.py | 0 .../{ => consumer_system}/consumer_system.py | 0 .../consumer_system/consumer_system_dto.py | 269 +++++ .../electricity_consumer/__init__.py | 0 .../electricity_consumer.py | 94 ++ .../fuel_consumer/__init__.py | 0 .../fuel_consumer/fuel_consumer.py | 121 ++ .../energy_components/fuel_model/__init__.py | 0 .../fuel_model/fuel_model.py | 85 ++ .../generator_set/generator_set_dto.py | 173 +++ .../installation/__init__.py | 0 .../installation/installation.py | 106 ++ .../energy_components/pump/component_dto.py | 31 + .../infrastructure/energy_components/utils.py | 26 + .../presentation/exporter/infrastructure.py | 2 +- .../presentation/json_result/mapper.py | 4 +- .../yaml/mappers/component_mapper.py | 12 +- .../components/system/yaml_consumer_system.py | 3 +- .../yaml_types/components/train/yaml_train.py | 2 +- .../yaml_types/components/yaml_compressor.py | 2 +- .../yaml/yaml_types/components/yaml_pump.py | 2 +- tests/libecalc/core/consumers/conftest.py | 20 +- .../core/consumers/test_consumer_utils.py | 4 +- .../libecalc/core/consumers/test_crossover.py | 2 +- tests/libecalc/core/models/test_fuel_model.py | 2 +- .../libecalc/dto/test_electricity_consumer.py | 6 +- tests/libecalc/dto/test_fuel_consumer.py | 12 +- tests/libecalc/dto/test_generator_set.py | 17 +- .../input/mappers/test_model_mapper.py | 5 +- .../libecalc/output/flow_diagram/conftest.py | 21 +- .../flow_diagram/test_ecalc_model_mapper.py | 6 +- .../json_result/test_aggregators.py | 15 +- 39 files changed, 1249 insertions(+), 1086 deletions(-) delete mode 100644 src/libecalc/domain/infrastructure/components.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/asset/__init__.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/asset/asset.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/base/component_dto.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/common.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/compressor/component_dto.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/consumer_system/__init__.py rename src/libecalc/domain/infrastructure/energy_components/{ => consumer_system}/consumer_system.py (100%) create mode 100644 src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system_dto.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/electricity_consumer/__init__.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/electricity_consumer/electricity_consumer.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/fuel_consumer/__init__.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/fuel_consumer/fuel_consumer.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/fuel_model/__init__.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/fuel_model/fuel_model.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/generator_set/generator_set_dto.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/installation/__init__.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/installation/installation.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/pump/component_dto.py create mode 100644 src/libecalc/domain/infrastructure/energy_components/utils.py diff --git a/src/libecalc/domain/infrastructure/__init__.py b/src/libecalc/domain/infrastructure/__init__.py index 89c600c4a8..72d305578f 100644 --- a/src/libecalc/domain/infrastructure/__init__.py +++ b/src/libecalc/domain/infrastructure/__init__.py @@ -1,8 +1,8 @@ -from libecalc.domain.infrastructure.components import ( - Asset, - BaseConsumer, +from libecalc.domain.infrastructure.energy_components.asset.asset import Asset +from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseConsumer +from libecalc.domain.infrastructure.energy_components.electricity_consumer.electricity_consumer import ( ElectricityConsumer, - FuelConsumer, - GeneratorSet, - Installation, ) +from libecalc.domain.infrastructure.energy_components.fuel_consumer.fuel_consumer import FuelConsumer +from libecalc.domain.infrastructure.energy_components.generator_set.generator_set_dto import GeneratorSet +from libecalc.domain.infrastructure.energy_components.installation.installation import Installation diff --git a/src/libecalc/domain/infrastructure/components.py b/src/libecalc/domain/infrastructure/components.py deleted file mode 100644 index 13dd458987..0000000000 --- a/src/libecalc/domain/infrastructure/components.py +++ /dev/null @@ -1,1012 +0,0 @@ -from abc import ABC, abstractmethod -from collections import defaultdict -from datetime import datetime -from typing import Annotated, Any, Literal, Optional, TypeVar, Union, overload - -import numpy as np -from pydantic import ConfigDict, Field, field_validator, model_validator -from pydantic_core.core_schema import ValidationInfo - -from libecalc.application.energy.component_energy_context import ComponentEnergyContext -from libecalc.application.energy.emitter import Emitter -from libecalc.application.energy.energy_component import EnergyComponent -from libecalc.application.energy.energy_model import EnergyModel -from libecalc.common.component_type import ComponentType -from libecalc.common.consumption_type import ConsumptionType -from libecalc.common.energy_usage_type import EnergyUsageType -from libecalc.common.logger import logger -from libecalc.common.priorities import Priorities -from libecalc.common.priority_optimizer import PriorityOptimizer -from libecalc.common.stream_conditions import TimeSeriesStreamConditions -from libecalc.common.string.string_utils import generate_id -from libecalc.common.temporal_model import TemporalModel -from libecalc.common.time_utils import Period, Periods -from libecalc.common.units import Unit -from libecalc.common.utils.rates import ( - RateType, - TimeSeriesFloat, - TimeSeriesInt, - TimeSeriesStreamDayRate, - TimeSeriesString, -) -from libecalc.common.variables import ExpressionEvaluator -from libecalc.core.models.compressor import create_compressor_model -from libecalc.core.models.generator import GeneratorModelSampled -from libecalc.core.models.pump import create_pump_model -from libecalc.core.result import ComponentResult, EcalcModelResult -from libecalc.core.result.emission import EmissionResult -from libecalc.domain.infrastructure.energy_components.compressor import Compressor -from libecalc.domain.infrastructure.energy_components.consumer_system import ( - ConsumerSystem as ConsumerSystemEnergyComponent, -) -from libecalc.domain.infrastructure.energy_components.generator_set.generator_set import Genset -from libecalc.domain.infrastructure.energy_components.legacy_consumer.component import ( - Consumer as ConsumerEnergyComponent, -) -from libecalc.domain.infrastructure.energy_components.legacy_consumer.consumer_function_mapper import EnergyModelMapper -from libecalc.domain.infrastructure.energy_components.pump import Pump -from libecalc.dto.base import ( - EcalcBaseModel, -) -from libecalc.dto.component_graph import ComponentGraph -from libecalc.dto.fuel_type import FuelType -from libecalc.dto.models import ( - ConsumerFunction, - ElectricEnergyUsageModel, - FuelEnergyUsageModel, - GeneratorSetSampled, -) -from libecalc.dto.models.compressor import CompressorModel -from libecalc.dto.models.pump import PumpModel -from libecalc.dto.types import ConsumerUserDefinedCategoryType, InstallationUserDefinedCategoryType -from libecalc.dto.utils.validators import ( - ComponentNameStr, - ExpressionType, - convert_expression, - validate_temporal_model, -) -from libecalc.expression import Expression -from libecalc.presentation.yaml.ltp_validation import ( - validate_generator_set_power_from_shore, -) -from libecalc.presentation.yaml.yaml_keywords import EcalcYamlKeywords -from libecalc.presentation.yaml.yaml_types.emitters.yaml_venting_emitter import ( - YamlVentingEmitter, -) - - -def check_model_energy_usage_type(model_data: dict[Period, ConsumerFunction], energy_type: EnergyUsageType): - for model in model_data.values(): - if model.energy_usage_type != energy_type: - raise ValueError(f"Model does not consume {energy_type.value}") - return model_data - - -class Component(EcalcBaseModel, ABC): - component_type: ComponentType - - @property - @abstractmethod - def id(self) -> str: ... - - -class BaseComponent(Component, ABC): - name: ComponentNameStr - - regularity: dict[Period, Expression] - - _validate_base_temporal_model = field_validator("regularity")(validate_temporal_model) - - @field_validator("regularity", mode="before") - @classmethod - def check_regularity(cls, regularity): - if isinstance(regularity, dict) and len(regularity.values()) > 0: - regularity = _convert_keys_in_dictionary_from_str_to_periods(regularity) - return regularity - - -class BaseEquipment(BaseComponent, ABC): - user_defined_category: dict[Period, ConsumerUserDefinedCategoryType] = Field(..., validate_default=True) - - @property - def id(self) -> str: - return generate_id(self.name) - - @field_validator("user_defined_category", mode="before") - def check_user_defined_category(cls, user_defined_category, info: ValidationInfo): - """Provide which value and context to make it easier for user to correct wrt mandatory changes.""" - if isinstance(user_defined_category, dict) and len(user_defined_category.values()) > 0: - user_defined_category = _convert_keys_in_dictionary_from_str_to_periods(user_defined_category) - for user_category in user_defined_category.values(): - if user_category not in list(ConsumerUserDefinedCategoryType): - name_context_str = "" - if (name := info.data.get("name")) is not None: - name_context_str = f"with the name {name}" - - raise ValueError( - f"CATEGORY: {user_category} is not allowed for {cls.__name__} {name_context_str}. Valid categories are: {[(consumer_user_defined_category.value) for consumer_user_defined_category in ConsumerUserDefinedCategoryType]}" - ) - - return user_defined_category - - -class BaseConsumer(BaseEquipment, ABC): - """Base class for all consumers.""" - - consumes: ConsumptionType - fuel: Optional[dict[Period, FuelType]] = None - - @field_validator("fuel", mode="before") - @classmethod - def validate_fuel_exist(cls, fuel, info: ValidationInfo): - """ - Make sure fuel is set if consumption type is FUEL. - """ - if isinstance(fuel, dict) and len(fuel.values()) > 0: - fuel = _convert_keys_in_dictionary_from_str_to_periods(fuel) - if info.data.get("consumes") == ConsumptionType.FUEL and (fuel is None or len(fuel) < 1): - msg = f"Missing fuel for fuel consumer '{info.data.get('name')}'" - raise ValueError(msg) - return fuel - - -class ElectricityConsumer(BaseConsumer, EnergyComponent): - component_type: Literal[ - ComponentType.COMPRESSOR, - ComponentType.PUMP, - ComponentType.GENERIC, - ComponentType.PUMP_SYSTEM, - ComponentType.COMPRESSOR_SYSTEM, - ] - consumes: Literal[ConsumptionType.ELECTRICITY] = ConsumptionType.ELECTRICITY - energy_usage_model: dict[ - Period, - ElectricEnergyUsageModel, - ] - - _validate_el_consumer_temporal_model = field_validator("energy_usage_model")(validate_temporal_model) - - _check_model_energy_usage = field_validator("energy_usage_model")( - lambda data: check_model_energy_usage_type(data, EnergyUsageType.POWER) - ) - - def is_fuel_consumer(self) -> bool: - return False - - def is_electricity_consumer(self) -> bool: - return True - - def is_provider(self) -> bool: - return False - - def is_container(self) -> bool: - return False - - def get_component_process_type(self) -> ComponentType: - return self.component_type - - def get_name(self) -> str: - return self.name - - def evaluate_energy_usage( - self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext - ) -> dict[str, EcalcModelResult]: - consumer_results: dict[str, EcalcModelResult] = {} - consumer = ConsumerEnergyComponent( - id=self.id, - name=self.name, - component_type=self.component_type, - regularity=TemporalModel(self.regularity), - consumes=self.consumes, - energy_usage_model=TemporalModel( - { - period: EnergyModelMapper.from_dto_to_domain(model) - for period, model in self.energy_usage_model.items() - } - ), - ) - consumer_results[self.id] = consumer.evaluate(expression_evaluator=expression_evaluator) - - return consumer_results - - @field_validator("energy_usage_model", mode="before") - @classmethod - def check_energy_usage_model(cls, energy_usage_model): - """ - Make sure that temporal models are converted to Period objects if they are strings - """ - if isinstance(energy_usage_model, dict) and len(energy_usage_model.values()) > 0: - energy_usage_model = _convert_keys_in_dictionary_from_str_to_periods(energy_usage_model) - return energy_usage_model - - -class FuelConsumer(BaseConsumer, Emitter, EnergyComponent): - component_type: Literal[ - ComponentType.COMPRESSOR, - ComponentType.GENERIC, - ComponentType.COMPRESSOR_SYSTEM, - ] - consumes: Literal[ConsumptionType.FUEL] = ConsumptionType.FUEL - fuel: dict[Period, FuelType] - energy_usage_model: dict[Period, FuelEnergyUsageModel] - - _validate_fuel_consumer_temporal_models = field_validator("energy_usage_model", "fuel")(validate_temporal_model) - _check_model_energy_usage = field_validator("energy_usage_model")( - lambda data: check_model_energy_usage_type(data, EnergyUsageType.FUEL) - ) - - def is_fuel_consumer(self) -> bool: - return True - - def is_electricity_consumer(self) -> bool: - return False - - def is_provider(self) -> bool: - return False - - def is_container(self) -> bool: - return False - - def get_component_process_type(self) -> ComponentType: - return self.component_type - - def get_name(self) -> str: - return self.name - - def evaluate_energy_usage( - self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext - ) -> dict[str, EcalcModelResult]: - consumer_results: dict[str, EcalcModelResult] = {} - consumer = ConsumerEnergyComponent( - id=self.id, - name=self.name, - component_type=self.component_type, - regularity=TemporalModel(self.regularity), - consumes=self.consumes, - energy_usage_model=TemporalModel( - { - period: EnergyModelMapper.from_dto_to_domain(model) - for period, model in self.energy_usage_model.items() - } - ), - ) - consumer_results[self.id] = consumer.evaluate(expression_evaluator=expression_evaluator) - - return consumer_results - - def evaluate_emissions( - self, - energy_context: ComponentEnergyContext, - energy_model: EnergyModel, - expression_evaluator: ExpressionEvaluator, - ) -> Optional[dict[str, EmissionResult]]: - fuel_model = FuelModel(self.fuel) - fuel_usage = energy_context.get_fuel_usage() - - assert fuel_usage is not None - - return fuel_model.evaluate_emissions( - expression_evaluator=expression_evaluator, - fuel_rate=fuel_usage.values, - ) - - @field_validator("energy_usage_model", mode="before") - @classmethod - def check_energy_usage_model(cls, energy_usage_model, info: ValidationInfo): - """ - Make sure that temporal models are converted to Period objects if they are strings - """ - if isinstance(energy_usage_model, dict) and len(energy_usage_model.values()) > 0: - energy_usage_model = _convert_keys_in_dictionary_from_str_to_periods(energy_usage_model) - return energy_usage_model - - @field_validator("fuel", mode="before") - @classmethod - def check_fuel(cls, fuel): - """ - Make sure that temporal models are converted to Period objects if they are strings - """ - if isinstance(fuel, dict) and len(fuel.values()) > 0: - fuel = _convert_keys_in_dictionary_from_str_to_periods(fuel) - return fuel - - -Consumer = Annotated[Union[FuelConsumer, ElectricityConsumer], Field(discriminator="consumes")] - - -class CompressorOperationalSettings(EcalcBaseModel): - rate: Expression - inlet_pressure: Expression - outlet_pressure: Expression - - -class PumpOperationalSettings(EcalcBaseModel): - rate: Expression - inlet_pressure: Expression - outlet_pressure: Expression - fluid_density: Expression - - -class CompressorComponent(BaseConsumer, EnergyComponent): - component_type: Literal[ComponentType.COMPRESSOR] = ComponentType.COMPRESSOR - energy_usage_model: dict[Period, CompressorModel] - - def is_fuel_consumer(self) -> bool: - return self.consumes == ConsumptionType.FUEL - - def is_electricity_consumer(self) -> bool: - return self.consumes == ConsumptionType.ELECTRICITY - - def is_provider(self) -> bool: - return False - - def is_container(self) -> bool: - return False - - def get_component_process_type(self) -> ComponentType: - return self.component_type - - def get_name(self) -> str: - return self.name - - -class PumpComponent(BaseConsumer, EnergyComponent): - component_type: Literal[ComponentType.PUMP] = ComponentType.PUMP - energy_usage_model: dict[Period, PumpModel] - - def is_fuel_consumer(self) -> bool: - return self.consumes == ConsumptionType.FUEL - - def is_electricity_consumer(self) -> bool: - return self.consumes == ConsumptionType.ELECTRICITY - - def is_provider(self) -> bool: - return False - - def is_container(self) -> bool: - return False - - def get_component_process_type(self) -> ComponentType: - return self.component_type - - def get_name(self) -> str: - return self.name - - -class Stream(EcalcBaseModel): - model_config = ConfigDict(populate_by_name=True) - - stream_name: Optional[str] = Field(None) - from_component_id: str - to_component_id: str - - -ConsumerComponent = TypeVar("ConsumerComponent", bound=Union[CompressorComponent, PumpComponent]) - - -class TrainComponent(BaseConsumer): - component_type: Literal[ComponentType.TRAIN_V2] = Field( - ComponentType.TRAIN_V2, - title="TYPE", - description="The type of the component", - alias="TYPE", - ) - stages: list[ConsumerComponent] - streams: list[Stream] - - -class ExpressionTimeSeries(EcalcBaseModel): - value: ExpressionType - unit: Unit - type: Optional[RateType] = None - - -class ExpressionStreamConditions(EcalcBaseModel): - rate: Optional[ExpressionTimeSeries] = None - pressure: Optional[ExpressionTimeSeries] = None - temperature: Optional[ExpressionTimeSeries] = None - fluid_density: Optional[ExpressionTimeSeries] = None - - -ConsumerID = str -PriorityID = str -StreamID = str - -SystemStreamConditions = dict[ConsumerID, dict[StreamID, ExpressionStreamConditions]] - - -class Crossover(EcalcBaseModel): - model_config = ConfigDict(populate_by_name=True) - - stream_name: Optional[str] = Field(None) - from_component_id: str - to_component_id: str - - -class SystemComponentConditions(EcalcBaseModel): - crossover: list[Crossover] - - -class ConsumerSystem(BaseConsumer, Emitter, EnergyComponent): - component_type: Literal[ComponentType.CONSUMER_SYSTEM_V2] = Field( - ComponentType.CONSUMER_SYSTEM_V2, - title="TYPE", - description="The type of the component", - ) - - component_conditions: SystemComponentConditions - stream_conditions_priorities: Priorities[SystemStreamConditions] - consumers: Union[list[CompressorComponent], list[PumpComponent]] - - def is_fuel_consumer(self) -> bool: - return self.consumes == ConsumptionType.FUEL - - def is_electricity_consumer(self) -> bool: - return self.consumes == ConsumptionType.ELECTRICITY - - def is_provider(self) -> bool: - return False - - def is_container(self) -> bool: - return True - - def get_component_process_type(self) -> ComponentType: - return self.component_type - - def get_name(self) -> str: - return self.name - - def evaluate_energy_usage( - self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext - ) -> dict[str, EcalcModelResult]: - consumer_results = {} - evaluated_stream_conditions = self.evaluate_stream_conditions( - expression_evaluator=expression_evaluator, - ) - optimizer = PriorityOptimizer() - - results_per_period: dict[str, dict[Period, ComponentResult]] = defaultdict(dict) - priorities_used = [] - for period in expression_evaluator.get_periods(): - consumers_for_period = [ - create_consumer( - consumer=consumer, - period=period, - ) - for consumer in self.consumers - ] - - consumer_system = ConsumerSystemEnergyComponent( - id=self.id, - consumers=consumers_for_period, - component_conditions=self.component_conditions, - ) - - def evaluator(priority: PriorityID): - stream_conditions_for_priority = evaluated_stream_conditions[priority] - stream_conditions_for_timestep = { - component_id: [stream_condition.for_period(period) for stream_condition in stream_conditions] - for component_id, stream_conditions in stream_conditions_for_priority.items() - } - return consumer_system.evaluate_consumers(stream_conditions_for_timestep) - - optimizer_result = optimizer.optimize( - priorities=list(evaluated_stream_conditions.keys()), - evaluator=evaluator, - ) - priorities_used.append(optimizer_result.priority_used) - for consumer_result in optimizer_result.priority_results: - results_per_period[consumer_result.id][period] = consumer_result - - priorities_used = TimeSeriesString( - periods=expression_evaluator.get_periods(), - values=priorities_used, - unit=Unit.NONE, - ) - # merge consumer results - consumer_ids = [consumer.id for consumer in self.consumers] - merged_consumer_results = [] - for consumer_id in consumer_ids: - first_result, *rest_results = list(results_per_period[consumer_id].values()) - merged_consumer_results.append(first_result.merge(*rest_results)) - - # Convert to legacy compatible operational_settings_used - priorities_to_int_map = { - priority_name: index + 1 for index, priority_name in enumerate(evaluated_stream_conditions.keys()) - } - operational_settings_used = TimeSeriesInt( - periods=priorities_used.periods, - values=[priorities_to_int_map[priority_name] for priority_name in priorities_used.values], - unit=priorities_used.unit, - ) - - system_result = ConsumerSystemEnergyComponent.get_system_result( - id=self.id, - consumer_results=merged_consumer_results, - operational_settings_used=operational_settings_used, - ) - consumer_results[self.id] = system_result - for consumer_result in merged_consumer_results: - consumer_results[consumer_result.id] = EcalcModelResult( - component_result=consumer_result, - sub_components=[], - models=[], - ) - - return consumer_results - - def evaluate_emissions( - self, - energy_context: ComponentEnergyContext, - energy_model: EnergyModel, - expression_evaluator: ExpressionEvaluator, - ) -> Optional[dict[str, EmissionResult]]: - if self.is_fuel_consumer(): - assert self.fuel is not None - fuel_model = FuelModel(self.fuel) - fuel_usage = energy_context.get_fuel_usage() - - assert fuel_usage is not None - - return fuel_model.evaluate_emissions( - expression_evaluator=expression_evaluator, - fuel_rate=fuel_usage.values, - ) - - def get_graph(self) -> ComponentGraph: - graph = ComponentGraph() - graph.add_node(self) - for consumer in self.consumers: - graph.add_node(consumer) - graph.add_edge(self.id, consumer.id) - return graph - - def evaluate_stream_conditions( - self, expression_evaluator: ExpressionEvaluator - ) -> Priorities[dict[ConsumerID, list[TimeSeriesStreamConditions]]]: - parsed_priorities: Priorities[dict[ConsumerID, list[TimeSeriesStreamConditions]]] = defaultdict(dict) - for priority_name, priority in self.stream_conditions_priorities.items(): - for consumer_name, streams_conditions in priority.items(): - parsed_priorities[priority_name][generate_id(consumer_name)] = [ - TimeSeriesStreamConditions( - id=generate_id(consumer_name, stream_name), - name="-".join([consumer_name, stream_name]), - rate=TimeSeriesStreamDayRate( - periods=expression_evaluator.get_periods(), - values=list( - expression_evaluator.evaluate( - Expression.setup_from_expression(stream_conditions.rate.value) - ) - ), - unit=stream_conditions.rate.unit, - ) - if stream_conditions.rate is not None - else None, - pressure=TimeSeriesFloat( - periods=expression_evaluator.get_periods(), - values=list( - expression_evaluator.evaluate( - expression=Expression.setup_from_expression(stream_conditions.pressure.value) - ) - ), - unit=stream_conditions.pressure.unit, - ) - if stream_conditions.pressure is not None - else None, - fluid_density=TimeSeriesFloat( - periods=expression_evaluator.get_periods(), - values=list( - expression_evaluator.evaluate( - expression=Expression.setup_from_expression(stream_conditions.fluid_density.value) - ) - ), - unit=stream_conditions.fluid_density.unit, - ) - if stream_conditions.fluid_density is not None - else None, - ) - for stream_name, stream_conditions in streams_conditions.items() - ] - return dict(parsed_priorities) - - -class GeneratorSet(BaseEquipment, Emitter, EnergyComponent): - component_type: Literal[ComponentType.GENERATOR_SET] = ComponentType.GENERATOR_SET - fuel: dict[Period, FuelType] - generator_set_model: dict[Period, GeneratorSetSampled] - consumers: list[ - Annotated[ - Union[ElectricityConsumer, ConsumerSystem], - Field(discriminator="component_type"), - ] - ] = Field(default_factory=list) - cable_loss: Optional[ExpressionType] = Field( - None, - title="CABLE_LOSS", - description="Power loss in cables from shore. " "Used to calculate onshore delivery/power supply onshore.", - ) - max_usage_from_shore: Optional[ExpressionType] = Field( - None, title="MAX_USAGE_FROM_SHORE", description="The peak load/effect that is expected for one hour, per year." - ) - - def is_fuel_consumer(self) -> bool: - return True - - def is_electricity_consumer(self) -> bool: - return False - - def is_provider(self) -> bool: - return True - - def is_container(self) -> bool: - return False - - def get_component_process_type(self) -> ComponentType: - return self.component_type - - def get_name(self) -> str: - return self.name - - def evaluate_energy_usage( - self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext - ) -> dict[str, EcalcModelResult]: - consumer_results: dict[str, EcalcModelResult] = {} - fuel_consumer = Genset( - id=self.id, - name=self.name, - temporal_generator_set_model=TemporalModel( - { - period: GeneratorModelSampled( - fuel_values=model.fuel_values, - power_values=model.power_values, - energy_usage_adjustment_constant=model.energy_usage_adjustment_constant, - energy_usage_adjustment_factor=model.energy_usage_adjustment_factor, - ) - for period, model in self.generator_set_model.items() - } - ), - ) - - consumer_results[self.id] = EcalcModelResult( - component_result=fuel_consumer.evaluate( - expression_evaluator=expression_evaluator, - power_requirement=context.get_power_requirement(), - ), - models=[], - sub_components=[], - ) - - return consumer_results - - def evaluate_emissions( - self, - energy_context: ComponentEnergyContext, - energy_model: EnergyModel, - expression_evaluator: ExpressionEvaluator, - ) -> Optional[dict[str, EmissionResult]]: - fuel_model = FuelModel(self.fuel) - fuel_usage = energy_context.get_fuel_usage() - - assert fuel_usage is not None - - return fuel_model.evaluate_emissions( - expression_evaluator=expression_evaluator, - fuel_rate=fuel_usage.values, - ) - - _validate_genset_temporal_models = field_validator("generator_set_model", "fuel")(validate_temporal_model) - - @field_validator("user_defined_category", mode="before") - @classmethod - def check_mandatory_category_for_generator_set(cls, user_defined_category, info: ValidationInfo): - """This could be handled automatically with Pydantic, but I want to inform the users in a better way, in - particular since we introduced a breaking change for this to be mandatory for GeneratorSets in v7.2. - """ - if user_defined_category is None or user_defined_category == "": - raise ValueError(f"CATEGORY is mandatory and must be set for '{info.data.get('name', cls.__name__)}'") - - return user_defined_category - - @field_validator("generator_set_model", mode="before") - @classmethod - def check_generator_set_model(cls, generator_set_model, info: ValidationInfo): - if isinstance(generator_set_model, dict) and len(generator_set_model.values()) > 0: - generator_set_model = _convert_keys_in_dictionary_from_str_to_periods(generator_set_model) - return generator_set_model - - @field_validator("fuel", mode="before") - @classmethod - def check_fuel(cls, fuel, info: ValidationInfo): - """ - Make sure that temporal models are converted to Period objects if they are strings - """ - if isinstance(fuel, dict) and len(fuel.values()) > 0: - fuel = _convert_keys_in_dictionary_from_str_to_periods(fuel) - return fuel - - @model_validator(mode="after") - def check_power_from_shore(self): - _check_power_from_shore_attributes = validate_generator_set_power_from_shore( - cable_loss=self.cable_loss, - max_usage_from_shore=self.max_usage_from_shore, - model_fields=self.model_fields, - category=self.user_defined_category, - ) - - return self - - def get_graph(self) -> ComponentGraph: - graph = ComponentGraph() - graph.add_node(self) - for electricity_consumer in self.consumers: - if hasattr(electricity_consumer, "get_graph"): - graph.add_subgraph(electricity_consumer.get_graph()) - else: - graph.add_node(electricity_consumer) - - graph.add_edge(self.id, electricity_consumer.id) - - return graph - - -class Installation(BaseComponent, EnergyComponent): - component_type: Literal[ComponentType.INSTALLATION] = ComponentType.INSTALLATION - - user_defined_category: Optional[InstallationUserDefinedCategoryType] = Field(default=None, validate_default=True) - hydrocarbon_export: dict[Period, Expression] - fuel_consumers: list[ - Annotated[ - Union[GeneratorSet, FuelConsumer, ConsumerSystem], - Field(discriminator="component_type"), - ] - ] = Field(default_factory=list) - venting_emitters: list[YamlVentingEmitter] = Field(default_factory=list) - - def is_fuel_consumer(self) -> bool: - return True - - def is_electricity_consumer(self) -> bool: - # Should maybe be True if power from shore? - return False - - def is_provider(self) -> bool: - return False - - def is_container(self) -> bool: - return True - - def get_component_process_type(self) -> ComponentType: - return self.component_type - - def get_name(self) -> str: - return self.name - - @property - def id(self) -> str: - return generate_id(self.name) - - _validate_installation_temporal_model = field_validator("hydrocarbon_export")(validate_temporal_model) - - _convert_expression_installation = field_validator("regularity", "hydrocarbon_export", mode="before")( - convert_expression - ) - - @field_validator("user_defined_category", mode="before") - def check_user_defined_category(cls, user_defined_category, info: ValidationInfo): - # Provide which value and context to make it easier for user to correct wrt mandatory changes. - if user_defined_category is not None: - if user_defined_category not in list(InstallationUserDefinedCategoryType): - name_context_str = "" - if (name := info.data.get("name")) is not None: - name_context_str = f"with the name {name}" - - raise ValueError( - f"CATEGORY: {user_defined_category} is not allowed for {cls.__name__} {name_context_str}. Valid categories are: {[str(installation_user_defined_category.value) for installation_user_defined_category in InstallationUserDefinedCategoryType]}" - ) - - return user_defined_category - - @model_validator(mode="after") - def check_fuel_consumers_or_venting_emitters_exist(self): - try: - if self.fuel_consumers or self.venting_emitters: - return self - except AttributeError: - raise ValueError( - f"Keywords are missing:\n It is required to specify at least one of the keywords " - f"{EcalcYamlKeywords.fuel_consumers}, {EcalcYamlKeywords.generator_sets} or {EcalcYamlKeywords.installation_venting_emitters} in the model.", - ) from None - - def get_graph(self) -> ComponentGraph: - graph = ComponentGraph() - graph.add_node(self) - for component in [*self.fuel_consumers, *self.venting_emitters]: - if hasattr(component, "get_graph"): - graph.add_subgraph(component.get_graph()) - else: - graph.add_node(component) - - graph.add_edge(self.id, component.id) - - return graph - - -class Asset(Component, EnergyComponent): - @property - def id(self): - return generate_id(self.name) - - name: ComponentNameStr - - installations: list[Installation] = Field(default_factory=list) - component_type: Literal[ComponentType.ASSET] = ComponentType.ASSET - - def is_fuel_consumer(self) -> bool: - return True - - def is_electricity_consumer(self) -> bool: - # Should maybe be True if power from shore? - return False - - def is_provider(self) -> bool: - return False - - def is_container(self) -> bool: - return True - - def get_component_process_type(self) -> ComponentType: - return self.component_type - - def get_name(self) -> str: - return self.name - - def get_graph(self) -> ComponentGraph: - graph = ComponentGraph() - graph.add_node(self) - for installation in self.installations: - graph.add_subgraph(installation.get_graph()) - graph.add_edge(self.id, installation.id) - - return graph - - -ComponentDTO = Union[ - Asset, - Installation, - GeneratorSet, - FuelConsumer, - ElectricityConsumer, - ConsumerSystem, - CompressorComponent, - PumpComponent, -] - - -def _convert_keys_in_dictionary_from_str_to_periods(data: dict[Union[str, Period], Any]) -> dict[Period, Any]: - if all(isinstance(key, str) for key in data.keys()): - return { - Period( - start=datetime.strptime(period.split(";")[0], "%Y-%m-%d %H:%M:%S"), - end=datetime.strptime(period.split(";")[1], "%Y-%m-%d %H:%M:%S"), - ): value - for period, value in data.items() - } - else: - return data - - -class FuelModel: - """A function to evaluate fuel related attributes for different time period - For each period, there is a data object with expressions for fuel related - attributes which may be evaluated for some variables and a fuel_rate. - """ - - def __init__(self, fuel_time_function_dict: dict[Period, FuelType]): - logger.debug("Creating fuel model") - self.temporal_fuel_model = fuel_time_function_dict - - def evaluate_emissions( - self, expression_evaluator: ExpressionEvaluator, fuel_rate: list[float] - ) -> dict[str, EmissionResult]: - """Evaluate fuel related expressions and results for a TimeSeriesCollection and a - fuel_rate array. - - First the fuel parameters are calculated by evaluating the fuel expressions and - the time_series object. - - Then the resulting emission volume is calculated based on the fuel rate: - - emission_rate = emission_factor * fuel_rate - - This is done per time period and all fuel related results both in terms of - fuel types and time periods, are merged into one common fuel collection results object. - - The length of the fuel_rate array must equal the length of the global list of periods. - It is assumed that the fuel_rate array origins from calculations based on the same time_series - object and thus will have the same length when used in this method. - """ - logger.debug("Evaluating fuel usage and emissions") - - fuel_rate = np.asarray(fuel_rate) - - # Creating a pseudo-default dict with all the emitters as keys. This is to handle changes in a temporal model. - emissions = { - emission_name: EmissionResult.create_empty(name=emission_name, periods=Periods([])) - for emission_name in { - emission.name for _, model in self.temporal_fuel_model.items() for emission in model.emissions - } - } - - for temporal_period, model in self.temporal_fuel_model.items(): - if Period.intersects(temporal_period, expression_evaluator.get_period()): - start_index, end_index = temporal_period.get_period_indices(expression_evaluator.get_periods()) - variables_map_this_period = expression_evaluator.get_subset( - start_index=start_index, - end_index=end_index, - ) - fuel_rate_this_period = fuel_rate[start_index:end_index] - for emission in model.emissions: - factor = variables_map_this_period.evaluate(expression=emission.factor) - - emission_rate_kg_per_day = fuel_rate_this_period * factor - emission_rate_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_rate_kg_per_day) - - result = EmissionResult( - name=emission.name, - periods=variables_map_this_period.get_periods(), - rate=TimeSeriesStreamDayRate( - periods=variables_map_this_period.get_periods(), - values=emission_rate_tons_per_day.tolist(), - unit=Unit.TONS_PER_DAY, - ), - ) - - emissions[emission.name].extend(result) - - for name in emissions: - if name not in [emission.name for emission in model.emissions]: - emissions[name].extend( - EmissionResult.create_empty(name=name, periods=variables_map_this_period.get_periods()) - ) - - return dict(sorted(emissions.items())) - - -@overload -def create_consumer(consumer: CompressorComponent, period: Period) -> Compressor: ... - - -@overload -def create_consumer(consumer: PumpComponent, period: Period) -> Pump: ... - - -def create_consumer( - consumer: Union[CompressorComponent, PumpComponent], - period: Period, -) -> Union[Compressor, Pump]: - periods = consumer.energy_usage_model.keys() - energy_usage_models = list(consumer.energy_usage_model.values()) - - model_for_period = None - for _period, energy_usage_model in zip(periods, energy_usage_models): - if period in _period: - model_for_period = energy_usage_model - - if model_for_period is None: - raise ValueError(f"Could not find model for consumer {consumer.name} at timestep {period}") - - if consumer.component_type == ComponentType.COMPRESSOR: - return Compressor( - id=consumer.id, - compressor_model=create_compressor_model( - compressor_model_dto=model_for_period, - ), - ) - elif consumer.component_type == ComponentType.PUMP: - return Pump( - id=consumer.id, - pump_model=create_pump_model( - pump_model_dto=model_for_period, - ), - ) - else: - raise TypeError(f"Unknown consumer. Received consumer with type '{consumer.component_type}'") diff --git a/src/libecalc/domain/infrastructure/energy_components/asset/__init__.py b/src/libecalc/domain/infrastructure/energy_components/asset/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/libecalc/domain/infrastructure/energy_components/asset/asset.py b/src/libecalc/domain/infrastructure/energy_components/asset/asset.py new file mode 100644 index 0000000000..20cc273052 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/asset/asset.py @@ -0,0 +1,50 @@ +from typing import Literal + +from pydantic import Field + +from libecalc.application.energy.energy_component import EnergyComponent +from libecalc.common.component_type import ComponentType +from libecalc.common.string.string_utils import generate_id +from libecalc.domain.infrastructure.energy_components.base.component_dto import Component +from libecalc.domain.infrastructure.energy_components.installation.installation import Installation +from libecalc.dto.component_graph import ComponentGraph +from libecalc.dto.utils.validators import ComponentNameStr + + +class Asset(Component, EnergyComponent): + @property + def id(self): + return generate_id(self.name) + + name: ComponentNameStr + + installations: list[Installation] = Field(default_factory=list) + component_type: Literal[ComponentType.ASSET] = ComponentType.ASSET + + def is_fuel_consumer(self) -> bool: + return True + + def is_electricity_consumer(self) -> bool: + # Should maybe be True if power from shore? + return False + + def is_provider(self) -> bool: + return False + + def is_container(self) -> bool: + return True + + def get_component_process_type(self) -> ComponentType: + return self.component_type + + def get_name(self) -> str: + return self.name + + def get_graph(self) -> ComponentGraph: + graph = ComponentGraph() + graph.add_node(self) + for installation in self.installations: + graph.add_subgraph(installation.get_graph()) + graph.add_edge(self.id, installation.id) + + return graph diff --git a/src/libecalc/domain/infrastructure/energy_components/base/component_dto.py b/src/libecalc/domain/infrastructure/energy_components/base/component_dto.py new file mode 100644 index 0000000000..67fe1a8eaa --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/base/component_dto.py @@ -0,0 +1,122 @@ +from abc import ABC, abstractmethod +from typing import Optional + +from pydantic import ConfigDict, Field, field_validator +from pydantic_core.core_schema import ValidationInfo + +from libecalc.common.component_type import ComponentType +from libecalc.common.consumption_type import ConsumptionType +from libecalc.common.string.string_utils import generate_id +from libecalc.common.time_utils import Period +from libecalc.common.units import Unit +from libecalc.common.utils.rates import RateType +from libecalc.domain.infrastructure.energy_components.utils import _convert_keys_in_dictionary_from_str_to_periods +from libecalc.dto.base import EcalcBaseModel +from libecalc.dto.fuel_type import FuelType +from libecalc.dto.types import ConsumerUserDefinedCategoryType +from libecalc.dto.utils.validators import ( + ComponentNameStr, + ExpressionType, + validate_temporal_model, +) +from libecalc.expression import Expression + + +class Component(EcalcBaseModel, ABC): + component_type: ComponentType + + @property + @abstractmethod + def id(self) -> str: ... + + +class BaseComponent(Component, ABC): + name: ComponentNameStr + + regularity: dict[Period, Expression] + + _validate_base_temporal_model = field_validator("regularity")(validate_temporal_model) + + @field_validator("regularity", mode="before") + @classmethod + def check_regularity(cls, regularity): + if isinstance(regularity, dict) and len(regularity.values()) > 0: + regularity = _convert_keys_in_dictionary_from_str_to_periods(regularity) + return regularity + + +class BaseEquipment(BaseComponent, ABC): + user_defined_category: dict[Period, ConsumerUserDefinedCategoryType] = Field(..., validate_default=True) + + @property + def id(self) -> str: + return generate_id(self.name) + + @field_validator("user_defined_category", mode="before") + def check_user_defined_category(cls, user_defined_category, info: ValidationInfo): + """Provide which value and context to make it easier for user to correct wrt mandatory changes.""" + if isinstance(user_defined_category, dict) and len(user_defined_category.values()) > 0: + user_defined_category = _convert_keys_in_dictionary_from_str_to_periods(user_defined_category) + for user_category in user_defined_category.values(): + if user_category not in list(ConsumerUserDefinedCategoryType): + name_context_str = "" + if (name := info.data.get("name")) is not None: + name_context_str = f"with the name {name}" + + raise ValueError( + f"CATEGORY: {user_category} is not allowed for {cls.__name__} {name_context_str}. Valid categories are: {[(consumer_user_defined_category.value) for consumer_user_defined_category in ConsumerUserDefinedCategoryType]}" + ) + + return user_defined_category + + +class BaseConsumer(BaseEquipment, ABC): + """Base class for all consumers.""" + + consumes: ConsumptionType + fuel: Optional[dict[Period, FuelType]] = None + + @field_validator("fuel", mode="before") + @classmethod + def validate_fuel_exist(cls, fuel, info: ValidationInfo): + """ + Make sure fuel is set if consumption type is FUEL. + """ + if isinstance(fuel, dict) and len(fuel.values()) > 0: + fuel = _convert_keys_in_dictionary_from_str_to_periods(fuel) + if info.data.get("consumes") == ConsumptionType.FUEL and (fuel is None or len(fuel) < 1): + msg = f"Missing fuel for fuel consumer '{info.data.get('name')}'" + raise ValueError(msg) + return fuel + + +class ExpressionTimeSeries(EcalcBaseModel): + value: ExpressionType + unit: Unit + type: Optional[RateType] = None + + +class ExpressionStreamConditions(EcalcBaseModel): + rate: Optional[ExpressionTimeSeries] = None + pressure: Optional[ExpressionTimeSeries] = None + temperature: Optional[ExpressionTimeSeries] = None + fluid_density: Optional[ExpressionTimeSeries] = None + + +ConsumerID = str +PriorityID = str +StreamID = str + +SystemStreamConditions = dict[ConsumerID, dict[StreamID, ExpressionStreamConditions]] + + +class Crossover(EcalcBaseModel): + model_config = ConfigDict(populate_by_name=True) + + stream_name: Optional[str] = Field(None) + from_component_id: str + to_component_id: str + + +class SystemComponentConditions(EcalcBaseModel): + crossover: list[Crossover] diff --git a/src/libecalc/domain/infrastructure/energy_components/common.py b/src/libecalc/domain/infrastructure/energy_components/common.py new file mode 100644 index 0000000000..fed67e8fd3 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/common.py @@ -0,0 +1,66 @@ +from typing import Annotated, Literal, Optional, TypeVar, Union + +from pydantic import ConfigDict, Field + +from libecalc.common.component_type import ComponentType +from libecalc.domain.infrastructure.energy_components.asset.asset import Asset +from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseConsumer +from libecalc.domain.infrastructure.energy_components.compressor.component_dto import CompressorComponent +from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system_dto import ConsumerSystem +from libecalc.domain.infrastructure.energy_components.electricity_consumer.electricity_consumer import ( + ElectricityConsumer, +) +from libecalc.domain.infrastructure.energy_components.fuel_consumer.fuel_consumer import FuelConsumer +from libecalc.domain.infrastructure.energy_components.generator_set.generator_set_dto import GeneratorSet +from libecalc.domain.infrastructure.energy_components.installation.installation import Installation +from libecalc.domain.infrastructure.energy_components.pump.component_dto import PumpComponent +from libecalc.dto.base import EcalcBaseModel +from libecalc.expression import Expression + +Consumer = Annotated[Union[FuelConsumer, ElectricityConsumer], Field(discriminator="consumes")] + +ComponentDTO = Union[ + Asset, + Installation, + GeneratorSet, + FuelConsumer, + ElectricityConsumer, + ConsumerSystem, + CompressorComponent, + PumpComponent, +] + + +class CompressorOperationalSettings(EcalcBaseModel): + rate: Expression + inlet_pressure: Expression + outlet_pressure: Expression + + +class PumpOperationalSettings(EcalcBaseModel): + rate: Expression + inlet_pressure: Expression + outlet_pressure: Expression + fluid_density: Expression + + +class Stream(EcalcBaseModel): + model_config = ConfigDict(populate_by_name=True) + + stream_name: Optional[str] = Field(None) + from_component_id: str + to_component_id: str + + +ConsumerComponent = TypeVar("ConsumerComponent", bound=Union[CompressorComponent, PumpComponent]) + + +class TrainComponent(BaseConsumer): + component_type: Literal[ComponentType.TRAIN_V2] = Field( + ComponentType.TRAIN_V2, + title="TYPE", + description="The type of the component", + alias="TYPE", + ) + stages: list[ConsumerComponent] + streams: list[Stream] diff --git a/src/libecalc/domain/infrastructure/energy_components/compressor/component_dto.py b/src/libecalc/domain/infrastructure/energy_components/compressor/component_dto.py new file mode 100644 index 0000000000..0935952051 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/compressor/component_dto.py @@ -0,0 +1,31 @@ +from typing import Literal + +from libecalc.application.energy.energy_component import EnergyComponent +from libecalc.common.component_type import ComponentType +from libecalc.common.consumption_type import ConsumptionType +from libecalc.common.time_utils import Period +from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseConsumer +from libecalc.dto.models.compressor import CompressorModel + + +class CompressorComponent(BaseConsumer, EnergyComponent): + component_type: Literal[ComponentType.COMPRESSOR] = ComponentType.COMPRESSOR + energy_usage_model: dict[Period, CompressorModel] + + def is_fuel_consumer(self) -> bool: + return self.consumes == ConsumptionType.FUEL + + def is_electricity_consumer(self) -> bool: + return self.consumes == ConsumptionType.ELECTRICITY + + def is_provider(self) -> bool: + return False + + def is_container(self) -> bool: + return False + + def get_component_process_type(self) -> ComponentType: + return self.component_type + + def get_name(self) -> str: + return self.name diff --git a/src/libecalc/domain/infrastructure/energy_components/consumer_system/__init__.py b/src/libecalc/domain/infrastructure/energy_components/consumer_system/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/libecalc/domain/infrastructure/energy_components/consumer_system.py b/src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system.py similarity index 100% rename from src/libecalc/domain/infrastructure/energy_components/consumer_system.py rename to src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system.py diff --git a/src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system_dto.py b/src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system_dto.py new file mode 100644 index 0000000000..bb1c850ed9 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system_dto.py @@ -0,0 +1,269 @@ +from collections import defaultdict +from typing import Literal, Optional, Union, overload + +from pydantic import Field + +from libecalc.application.energy.component_energy_context import ComponentEnergyContext +from libecalc.application.energy.emitter import Emitter +from libecalc.application.energy.energy_component import EnergyComponent +from libecalc.application.energy.energy_model import EnergyModel +from libecalc.common.component_type import ComponentType +from libecalc.common.consumption_type import ConsumptionType +from libecalc.common.priorities import Priorities +from libecalc.common.priority_optimizer import PriorityOptimizer +from libecalc.common.stream_conditions import TimeSeriesStreamConditions +from libecalc.common.string.string_utils import generate_id +from libecalc.common.time_utils import Period +from libecalc.common.units import Unit +from libecalc.common.utils.rates import ( + TimeSeriesFloat, + TimeSeriesInt, + TimeSeriesStreamDayRate, + TimeSeriesString, +) +from libecalc.common.variables import ExpressionEvaluator +from libecalc.core.models.compressor import create_compressor_model +from libecalc.core.models.pump import create_pump_model +from libecalc.core.result import ComponentResult, EcalcModelResult +from libecalc.core.result.emission import EmissionResult +from libecalc.domain.infrastructure.energy_components.base.component_dto import ( + BaseConsumer, + ConsumerID, + PriorityID, + SystemComponentConditions, + SystemStreamConditions, +) +from libecalc.domain.infrastructure.energy_components.compressor import Compressor +from libecalc.domain.infrastructure.energy_components.compressor.component_dto import CompressorComponent +from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system import ( + ConsumerSystem as ConsumerSystemEnergyComponent, +) +from libecalc.domain.infrastructure.energy_components.fuel_model.fuel_model import FuelModel +from libecalc.domain.infrastructure.energy_components.pump import Pump +from libecalc.domain.infrastructure.energy_components.pump.component_dto import PumpComponent +from libecalc.dto.component_graph import ComponentGraph +from libecalc.expression import Expression + + +class ConsumerSystem(BaseConsumer, Emitter, EnergyComponent): + component_type: Literal[ComponentType.CONSUMER_SYSTEM_V2] = Field( + ComponentType.CONSUMER_SYSTEM_V2, + title="TYPE", + description="The type of the component", + ) + + component_conditions: SystemComponentConditions + stream_conditions_priorities: Priorities[SystemStreamConditions] + consumers: Union[list[CompressorComponent], list[PumpComponent]] + + def is_fuel_consumer(self) -> bool: + return self.consumes == ConsumptionType.FUEL + + def is_electricity_consumer(self) -> bool: + return self.consumes == ConsumptionType.ELECTRICITY + + def is_provider(self) -> bool: + return False + + def is_container(self) -> bool: + return True + + def get_component_process_type(self) -> ComponentType: + return self.component_type + + def get_name(self) -> str: + return self.name + + def evaluate_energy_usage( + self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext + ) -> dict[str, EcalcModelResult]: + consumer_results = {} + evaluated_stream_conditions = self.evaluate_stream_conditions( + expression_evaluator=expression_evaluator, + ) + optimizer = PriorityOptimizer() + + results_per_period: dict[str, dict[Period, ComponentResult]] = defaultdict(dict) + priorities_used = [] + for period in expression_evaluator.get_periods(): + consumers_for_period = [ + create_consumer( + consumer=consumer, + period=period, + ) + for consumer in self.consumers + ] + + consumer_system = ConsumerSystemEnergyComponent( + id=self.id, + consumers=consumers_for_period, + component_conditions=self.component_conditions, + ) + + def evaluator(priority: PriorityID): + stream_conditions_for_priority = evaluated_stream_conditions[priority] + stream_conditions_for_timestep = { + component_id: [stream_condition.for_period(period) for stream_condition in stream_conditions] + for component_id, stream_conditions in stream_conditions_for_priority.items() + } + return consumer_system.evaluate_consumers(stream_conditions_for_timestep) + + optimizer_result = optimizer.optimize( + priorities=list(evaluated_stream_conditions.keys()), + evaluator=evaluator, + ) + priorities_used.append(optimizer_result.priority_used) + for consumer_result in optimizer_result.priority_results: + results_per_period[consumer_result.id][period] = consumer_result + + priorities_used = TimeSeriesString( + periods=expression_evaluator.get_periods(), + values=priorities_used, + unit=Unit.NONE, + ) + # merge consumer results + consumer_ids = [consumer.id for consumer in self.consumers] + merged_consumer_results = [] + for consumer_id in consumer_ids: + first_result, *rest_results = list(results_per_period[consumer_id].values()) + merged_consumer_results.append(first_result.merge(*rest_results)) + + # Convert to legacy compatible operational_settings_used + priorities_to_int_map = { + priority_name: index + 1 for index, priority_name in enumerate(evaluated_stream_conditions.keys()) + } + operational_settings_used = TimeSeriesInt( + periods=priorities_used.periods, + values=[priorities_to_int_map[priority_name] for priority_name in priorities_used.values], + unit=priorities_used.unit, + ) + + system_result = ConsumerSystemEnergyComponent.get_system_result( + id=self.id, + consumer_results=merged_consumer_results, + operational_settings_used=operational_settings_used, + ) + consumer_results[self.id] = system_result + for consumer_result in merged_consumer_results: + consumer_results[consumer_result.id] = EcalcModelResult( + component_result=consumer_result, + sub_components=[], + models=[], + ) + + return consumer_results + + def evaluate_emissions( + self, + energy_context: ComponentEnergyContext, + energy_model: EnergyModel, + expression_evaluator: ExpressionEvaluator, + ) -> Optional[dict[str, EmissionResult]]: + if self.is_fuel_consumer(): + assert self.fuel is not None + fuel_model = FuelModel(self.fuel) + fuel_usage = energy_context.get_fuel_usage() + + assert fuel_usage is not None + + return fuel_model.evaluate_emissions( + expression_evaluator=expression_evaluator, + fuel_rate=fuel_usage.values, + ) + + def get_graph(self) -> ComponentGraph: + graph = ComponentGraph() + graph.add_node(self) + for consumer in self.consumers: + graph.add_node(consumer) + graph.add_edge(self.id, consumer.id) + return graph + + def evaluate_stream_conditions( + self, expression_evaluator: ExpressionEvaluator + ) -> Priorities[dict[ConsumerID, list[TimeSeriesStreamConditions]]]: + parsed_priorities: Priorities[dict[ConsumerID, list[TimeSeriesStreamConditions]]] = defaultdict(dict) + for priority_name, priority in self.stream_conditions_priorities.items(): + for consumer_name, streams_conditions in priority.items(): + parsed_priorities[priority_name][generate_id(consumer_name)] = [ + TimeSeriesStreamConditions( + id=generate_id(consumer_name, stream_name), + name="-".join([consumer_name, stream_name]), + rate=TimeSeriesStreamDayRate( + periods=expression_evaluator.get_periods(), + values=list( + expression_evaluator.evaluate( + Expression.setup_from_expression(stream_conditions.rate.value) + ) + ), + unit=stream_conditions.rate.unit, + ) + if stream_conditions.rate is not None + else None, + pressure=TimeSeriesFloat( + periods=expression_evaluator.get_periods(), + values=list( + expression_evaluator.evaluate( + expression=Expression.setup_from_expression(stream_conditions.pressure.value) + ) + ), + unit=stream_conditions.pressure.unit, + ) + if stream_conditions.pressure is not None + else None, + fluid_density=TimeSeriesFloat( + periods=expression_evaluator.get_periods(), + values=list( + expression_evaluator.evaluate( + expression=Expression.setup_from_expression(stream_conditions.fluid_density.value) + ) + ), + unit=stream_conditions.fluid_density.unit, + ) + if stream_conditions.fluid_density is not None + else None, + ) + for stream_name, stream_conditions in streams_conditions.items() + ] + return dict(parsed_priorities) + + +@overload +def create_consumer(consumer: CompressorComponent, period: Period) -> Compressor: ... + + +@overload +def create_consumer(consumer: PumpComponent, period: Period) -> Pump: ... + + +def create_consumer( + consumer: Union[CompressorComponent, PumpComponent], + period: Period, +) -> Union[Compressor, Pump]: + periods = consumer.energy_usage_model.keys() + energy_usage_models = list(consumer.energy_usage_model.values()) + + model_for_period = None + for _period, energy_usage_model in zip(periods, energy_usage_models): + if period in _period: + model_for_period = energy_usage_model + + if model_for_period is None: + raise ValueError(f"Could not find model for consumer {consumer.name} at timestep {period}") + + if consumer.component_type == ComponentType.COMPRESSOR: + return Compressor( + id=consumer.id, + compressor_model=create_compressor_model( + compressor_model_dto=model_for_period, + ), + ) + elif consumer.component_type == ComponentType.PUMP: + return Pump( + id=consumer.id, + pump_model=create_pump_model( + pump_model_dto=model_for_period, + ), + ) + else: + raise TypeError(f"Unknown consumer. Received consumer with type '{consumer.component_type}'") diff --git a/src/libecalc/domain/infrastructure/energy_components/electricity_consumer/__init__.py b/src/libecalc/domain/infrastructure/energy_components/electricity_consumer/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/libecalc/domain/infrastructure/energy_components/electricity_consumer/electricity_consumer.py b/src/libecalc/domain/infrastructure/energy_components/electricity_consumer/electricity_consumer.py new file mode 100644 index 0000000000..0ea74a54ae --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/electricity_consumer/electricity_consumer.py @@ -0,0 +1,94 @@ +from typing import Literal + +from pydantic import field_validator + +from libecalc.application.energy.component_energy_context import ComponentEnergyContext +from libecalc.application.energy.energy_component import EnergyComponent +from libecalc.common.component_type import ComponentType +from libecalc.common.consumption_type import ConsumptionType +from libecalc.common.energy_usage_type import EnergyUsageType +from libecalc.common.temporal_model import TemporalModel +from libecalc.common.time_utils import Period +from libecalc.common.variables import ExpressionEvaluator +from libecalc.core.result import EcalcModelResult +from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseConsumer +from libecalc.domain.infrastructure.energy_components.legacy_consumer.component import ( + Consumer as ConsumerEnergyComponent, +) +from libecalc.domain.infrastructure.energy_components.legacy_consumer.consumer_function_mapper import EnergyModelMapper +from libecalc.domain.infrastructure.energy_components.utils import ( + _convert_keys_in_dictionary_from_str_to_periods, + check_model_energy_usage_type, +) +from libecalc.dto.models import ElectricEnergyUsageModel +from libecalc.dto.utils.validators import validate_temporal_model + + +class ElectricityConsumer(BaseConsumer, EnergyComponent): + component_type: Literal[ + ComponentType.COMPRESSOR, + ComponentType.PUMP, + ComponentType.GENERIC, + ComponentType.PUMP_SYSTEM, + ComponentType.COMPRESSOR_SYSTEM, + ] + consumes: Literal[ConsumptionType.ELECTRICITY] = ConsumptionType.ELECTRICITY + energy_usage_model: dict[ + Period, + ElectricEnergyUsageModel, + ] + + _validate_el_consumer_temporal_model = field_validator("energy_usage_model")(validate_temporal_model) + + _check_model_energy_usage = field_validator("energy_usage_model")( + lambda data: check_model_energy_usage_type(data, EnergyUsageType.POWER) + ) + + def is_fuel_consumer(self) -> bool: + return False + + def is_electricity_consumer(self) -> bool: + return True + + def is_provider(self) -> bool: + return False + + def is_container(self) -> bool: + return False + + def get_component_process_type(self) -> ComponentType: + return self.component_type + + def get_name(self) -> str: + return self.name + + def evaluate_energy_usage( + self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext + ) -> dict[str, EcalcModelResult]: + consumer_results: dict[str, EcalcModelResult] = {} + consumer = ConsumerEnergyComponent( + id=self.id, + name=self.name, + component_type=self.component_type, + regularity=TemporalModel(self.regularity), + consumes=self.consumes, + energy_usage_model=TemporalModel( + { + period: EnergyModelMapper.from_dto_to_domain(model) + for period, model in self.energy_usage_model.items() + } + ), + ) + consumer_results[self.id] = consumer.evaluate(expression_evaluator=expression_evaluator) + + return consumer_results + + @field_validator("energy_usage_model", mode="before") + @classmethod + def check_energy_usage_model(cls, energy_usage_model): + """ + Make sure that temporal models are converted to Period objects if they are strings + """ + if isinstance(energy_usage_model, dict) and len(energy_usage_model.values()) > 0: + energy_usage_model = _convert_keys_in_dictionary_from_str_to_periods(energy_usage_model) + return energy_usage_model diff --git a/src/libecalc/domain/infrastructure/energy_components/fuel_consumer/__init__.py b/src/libecalc/domain/infrastructure/energy_components/fuel_consumer/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/libecalc/domain/infrastructure/energy_components/fuel_consumer/fuel_consumer.py b/src/libecalc/domain/infrastructure/energy_components/fuel_consumer/fuel_consumer.py new file mode 100644 index 0000000000..cac74245d7 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/fuel_consumer/fuel_consumer.py @@ -0,0 +1,121 @@ +from typing import Literal, Optional + +from pydantic import field_validator +from pydantic_core.core_schema import ValidationInfo + +from libecalc.application.energy.component_energy_context import ComponentEnergyContext +from libecalc.application.energy.emitter import Emitter +from libecalc.application.energy.energy_component import EnergyComponent +from libecalc.application.energy.energy_model import EnergyModel +from libecalc.common.component_type import ComponentType +from libecalc.common.consumption_type import ConsumptionType +from libecalc.common.energy_usage_type import EnergyUsageType +from libecalc.common.temporal_model import TemporalModel +from libecalc.common.time_utils import Period +from libecalc.common.variables import ExpressionEvaluator +from libecalc.core.result import EcalcModelResult +from libecalc.core.result.emission import EmissionResult +from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseConsumer +from libecalc.domain.infrastructure.energy_components.fuel_model.fuel_model import FuelModel +from libecalc.domain.infrastructure.energy_components.legacy_consumer.component import ( + Consumer as ConsumerEnergyComponent, +) +from libecalc.domain.infrastructure.energy_components.legacy_consumer.consumer_function_mapper import EnergyModelMapper +from libecalc.domain.infrastructure.energy_components.utils import ( + _convert_keys_in_dictionary_from_str_to_periods, + check_model_energy_usage_type, +) +from libecalc.dto.fuel_type import FuelType +from libecalc.dto.models import FuelEnergyUsageModel +from libecalc.dto.utils.validators import validate_temporal_model + + +class FuelConsumer(BaseConsumer, Emitter, EnergyComponent): + component_type: Literal[ + ComponentType.COMPRESSOR, + ComponentType.GENERIC, + ComponentType.COMPRESSOR_SYSTEM, + ] + consumes: Literal[ConsumptionType.FUEL] = ConsumptionType.FUEL + fuel: dict[Period, FuelType] + energy_usage_model: dict[Period, FuelEnergyUsageModel] + + _validate_fuel_consumer_temporal_models = field_validator("energy_usage_model", "fuel")(validate_temporal_model) + _check_model_energy_usage = field_validator("energy_usage_model")( + lambda data: check_model_energy_usage_type(data, EnergyUsageType.FUEL) + ) + + def is_fuel_consumer(self) -> bool: + return True + + def is_electricity_consumer(self) -> bool: + return False + + def is_provider(self) -> bool: + return False + + def is_container(self) -> bool: + return False + + def get_component_process_type(self) -> ComponentType: + return self.component_type + + def get_name(self) -> str: + return self.name + + def evaluate_energy_usage( + self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext + ) -> dict[str, EcalcModelResult]: + consumer_results: dict[str, EcalcModelResult] = {} + consumer = ConsumerEnergyComponent( + id=self.id, + name=self.name, + component_type=self.component_type, + regularity=TemporalModel(self.regularity), + consumes=self.consumes, + energy_usage_model=TemporalModel( + { + period: EnergyModelMapper.from_dto_to_domain(model) + for period, model in self.energy_usage_model.items() + } + ), + ) + consumer_results[self.id] = consumer.evaluate(expression_evaluator=expression_evaluator) + + return consumer_results + + def evaluate_emissions( + self, + energy_context: ComponentEnergyContext, + energy_model: EnergyModel, + expression_evaluator: ExpressionEvaluator, + ) -> Optional[dict[str, EmissionResult]]: + fuel_model = FuelModel(self.fuel) + fuel_usage = energy_context.get_fuel_usage() + + assert fuel_usage is not None + + return fuel_model.evaluate_emissions( + expression_evaluator=expression_evaluator, + fuel_rate=fuel_usage.values, + ) + + @field_validator("energy_usage_model", mode="before") + @classmethod + def check_energy_usage_model(cls, energy_usage_model, info: ValidationInfo): + """ + Make sure that temporal models are converted to Period objects if they are strings + """ + if isinstance(energy_usage_model, dict) and len(energy_usage_model.values()) > 0: + energy_usage_model = _convert_keys_in_dictionary_from_str_to_periods(energy_usage_model) + return energy_usage_model + + @field_validator("fuel", mode="before") + @classmethod + def check_fuel(cls, fuel): + """ + Make sure that temporal models are converted to Period objects if they are strings + """ + if isinstance(fuel, dict) and len(fuel.values()) > 0: + fuel = _convert_keys_in_dictionary_from_str_to_periods(fuel) + return fuel diff --git a/src/libecalc/domain/infrastructure/energy_components/fuel_model/__init__.py b/src/libecalc/domain/infrastructure/energy_components/fuel_model/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/libecalc/domain/infrastructure/energy_components/fuel_model/fuel_model.py b/src/libecalc/domain/infrastructure/energy_components/fuel_model/fuel_model.py new file mode 100644 index 0000000000..559ba9dde9 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/fuel_model/fuel_model.py @@ -0,0 +1,85 @@ +import numpy as np + +from libecalc.common.logger import logger +from libecalc.common.time_utils import Period, Periods +from libecalc.common.units import Unit +from libecalc.common.utils.rates import TimeSeriesStreamDayRate +from libecalc.common.variables import ExpressionEvaluator +from libecalc.core.result.emission import EmissionResult +from libecalc.dto.fuel_type import FuelType + + +class FuelModel: + """A function to evaluate fuel related attributes for different time period + For each period, there is a data object with expressions for fuel related + attributes which may be evaluated for some variables and a fuel_rate. + """ + + def __init__(self, fuel_time_function_dict: dict[Period, FuelType]): + logger.debug("Creating fuel model") + self.temporal_fuel_model = fuel_time_function_dict + + def evaluate_emissions( + self, expression_evaluator: ExpressionEvaluator, fuel_rate: list[float] + ) -> dict[str, EmissionResult]: + """Evaluate fuel related expressions and results for a TimeSeriesCollection and a + fuel_rate array. + + First the fuel parameters are calculated by evaluating the fuel expressions and + the time_series object. + + Then the resulting emission volume is calculated based on the fuel rate: + - emission_rate = emission_factor * fuel_rate + + This is done per time period and all fuel related results both in terms of + fuel types and time periods, are merged into one common fuel collection results object. + + The length of the fuel_rate array must equal the length of the global list of periods. + It is assumed that the fuel_rate array origins from calculations based on the same time_series + object and thus will have the same length when used in this method. + """ + logger.debug("Evaluating fuel usage and emissions") + + fuel_rate = np.asarray(fuel_rate) + + # Creating a pseudo-default dict with all the emitters as keys. This is to handle changes in a temporal model. + emissions = { + emission_name: EmissionResult.create_empty(name=emission_name, periods=Periods([])) + for emission_name in { + emission.name for _, model in self.temporal_fuel_model.items() for emission in model.emissions + } + } + + for temporal_period, model in self.temporal_fuel_model.items(): + if Period.intersects(temporal_period, expression_evaluator.get_period()): + start_index, end_index = temporal_period.get_period_indices(expression_evaluator.get_periods()) + variables_map_this_period = expression_evaluator.get_subset( + start_index=start_index, + end_index=end_index, + ) + fuel_rate_this_period = fuel_rate[start_index:end_index] + for emission in model.emissions: + factor = variables_map_this_period.evaluate(expression=emission.factor) + + emission_rate_kg_per_day = fuel_rate_this_period * factor + emission_rate_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_rate_kg_per_day) + + result = EmissionResult( + name=emission.name, + periods=variables_map_this_period.get_periods(), + rate=TimeSeriesStreamDayRate( + periods=variables_map_this_period.get_periods(), + values=emission_rate_tons_per_day.tolist(), + unit=Unit.TONS_PER_DAY, + ), + ) + + emissions[emission.name].extend(result) + + for name in emissions: + if name not in [emission.name for emission in model.emissions]: + emissions[name].extend( + EmissionResult.create_empty(name=name, periods=variables_map_this_period.get_periods()) + ) + + return dict(sorted(emissions.items())) diff --git a/src/libecalc/domain/infrastructure/energy_components/generator_set/generator_set_dto.py b/src/libecalc/domain/infrastructure/energy_components/generator_set/generator_set_dto.py new file mode 100644 index 0000000000..441a1f7791 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/generator_set/generator_set_dto.py @@ -0,0 +1,173 @@ +from typing import Annotated, Literal, Optional, Union + +from pydantic import Field, field_validator, model_validator +from pydantic_core.core_schema import ValidationInfo + +from libecalc.application.energy.component_energy_context import ComponentEnergyContext +from libecalc.application.energy.emitter import Emitter +from libecalc.application.energy.energy_component import EnergyComponent +from libecalc.application.energy.energy_model import EnergyModel +from libecalc.common.component_type import ComponentType +from libecalc.common.temporal_model import TemporalModel +from libecalc.common.time_utils import Period +from libecalc.common.variables import ExpressionEvaluator +from libecalc.core.models.generator import GeneratorModelSampled +from libecalc.core.result import EcalcModelResult +from libecalc.core.result.emission import EmissionResult +from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseEquipment +from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system_dto import ConsumerSystem +from libecalc.domain.infrastructure.energy_components.electricity_consumer.electricity_consumer import ( + ElectricityConsumer, +) +from libecalc.domain.infrastructure.energy_components.fuel_model.fuel_model import FuelModel +from libecalc.domain.infrastructure.energy_components.generator_set.generator_set import Genset +from libecalc.domain.infrastructure.energy_components.utils import _convert_keys_in_dictionary_from_str_to_periods +from libecalc.dto.component_graph import ComponentGraph +from libecalc.dto.fuel_type import FuelType +from libecalc.dto.models import GeneratorSetSampled +from libecalc.dto.utils.validators import ( + ExpressionType, + validate_temporal_model, +) +from libecalc.presentation.yaml.ltp_validation import ( + validate_generator_set_power_from_shore, +) + + +class GeneratorSet(BaseEquipment, Emitter, EnergyComponent): + component_type: Literal[ComponentType.GENERATOR_SET] = ComponentType.GENERATOR_SET + fuel: dict[Period, FuelType] + generator_set_model: dict[Period, GeneratorSetSampled] + consumers: list[ + Annotated[ + Union[ElectricityConsumer, ConsumerSystem], + Field(discriminator="component_type"), + ] + ] = Field(default_factory=list) + cable_loss: Optional[ExpressionType] = Field( + None, + title="CABLE_LOSS", + description="Power loss in cables from shore. " "Used to calculate onshore delivery/power supply onshore.", + ) + max_usage_from_shore: Optional[ExpressionType] = Field( + None, title="MAX_USAGE_FROM_SHORE", description="The peak load/effect that is expected for one hour, per year." + ) + + def is_fuel_consumer(self) -> bool: + return True + + def is_electricity_consumer(self) -> bool: + return False + + def is_provider(self) -> bool: + return True + + def is_container(self) -> bool: + return False + + def get_component_process_type(self) -> ComponentType: + return self.component_type + + def get_name(self) -> str: + return self.name + + def evaluate_energy_usage( + self, expression_evaluator: ExpressionEvaluator, context: ComponentEnergyContext + ) -> dict[str, EcalcModelResult]: + consumer_results: dict[str, EcalcModelResult] = {} + fuel_consumer = Genset( + id=self.id, + name=self.name, + temporal_generator_set_model=TemporalModel( + { + period: GeneratorModelSampled( + fuel_values=model.fuel_values, + power_values=model.power_values, + energy_usage_adjustment_constant=model.energy_usage_adjustment_constant, + energy_usage_adjustment_factor=model.energy_usage_adjustment_factor, + ) + for period, model in self.generator_set_model.items() + } + ), + ) + + consumer_results[self.id] = EcalcModelResult( + component_result=fuel_consumer.evaluate( + expression_evaluator=expression_evaluator, + power_requirement=context.get_power_requirement(), + ), + models=[], + sub_components=[], + ) + + return consumer_results + + def evaluate_emissions( + self, + energy_context: ComponentEnergyContext, + energy_model: EnergyModel, + expression_evaluator: ExpressionEvaluator, + ) -> Optional[dict[str, EmissionResult]]: + fuel_model = FuelModel(self.fuel) + fuel_usage = energy_context.get_fuel_usage() + + assert fuel_usage is not None + + return fuel_model.evaluate_emissions( + expression_evaluator=expression_evaluator, + fuel_rate=fuel_usage.values, + ) + + _validate_genset_temporal_models = field_validator("generator_set_model", "fuel")(validate_temporal_model) + + @field_validator("user_defined_category", mode="before") + @classmethod + def check_mandatory_category_for_generator_set(cls, user_defined_category, info: ValidationInfo): + """This could be handled automatically with Pydantic, but I want to inform the users in a better way, in + particular since we introduced a breaking change for this to be mandatory for GeneratorSets in v7.2. + """ + if user_defined_category is None or user_defined_category == "": + raise ValueError(f"CATEGORY is mandatory and must be set for '{info.data.get('name', cls.__name__)}'") + + return user_defined_category + + @field_validator("generator_set_model", mode="before") + @classmethod + def check_generator_set_model(cls, generator_set_model, info: ValidationInfo): + if isinstance(generator_set_model, dict) and len(generator_set_model.values()) > 0: + generator_set_model = _convert_keys_in_dictionary_from_str_to_periods(generator_set_model) + return generator_set_model + + @field_validator("fuel", mode="before") + @classmethod + def check_fuel(cls, fuel, info: ValidationInfo): + """ + Make sure that temporal models are converted to Period objects if they are strings + """ + if isinstance(fuel, dict) and len(fuel.values()) > 0: + fuel = _convert_keys_in_dictionary_from_str_to_periods(fuel) + return fuel + + @model_validator(mode="after") + def check_power_from_shore(self): + _check_power_from_shore_attributes = validate_generator_set_power_from_shore( + cable_loss=self.cable_loss, + max_usage_from_shore=self.max_usage_from_shore, + model_fields=self.model_fields, + category=self.user_defined_category, + ) + + return self + + def get_graph(self) -> ComponentGraph: + graph = ComponentGraph() + graph.add_node(self) + for electricity_consumer in self.consumers: + if hasattr(electricity_consumer, "get_graph"): + graph.add_subgraph(electricity_consumer.get_graph()) + else: + graph.add_node(electricity_consumer) + + graph.add_edge(self.id, electricity_consumer.id) + + return graph diff --git a/src/libecalc/domain/infrastructure/energy_components/installation/__init__.py b/src/libecalc/domain/infrastructure/energy_components/installation/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/libecalc/domain/infrastructure/energy_components/installation/installation.py b/src/libecalc/domain/infrastructure/energy_components/installation/installation.py new file mode 100644 index 0000000000..8048a5b520 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/installation/installation.py @@ -0,0 +1,106 @@ +from typing import Annotated, Literal, Optional, Union + +from pydantic import Field, field_validator, model_validator +from pydantic_core.core_schema import ValidationInfo + +from libecalc.application.energy.energy_component import EnergyComponent +from libecalc.common.component_type import ComponentType +from libecalc.common.string.string_utils import generate_id +from libecalc.common.time_utils import Period +from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseComponent +from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system_dto import ConsumerSystem +from libecalc.domain.infrastructure.energy_components.fuel_consumer.fuel_consumer import FuelConsumer +from libecalc.domain.infrastructure.energy_components.generator_set.generator_set_dto import GeneratorSet +from libecalc.dto.component_graph import ComponentGraph +from libecalc.dto.types import InstallationUserDefinedCategoryType +from libecalc.dto.utils.validators import ( + convert_expression, + validate_temporal_model, +) +from libecalc.expression import Expression +from libecalc.presentation.yaml.yaml_keywords import EcalcYamlKeywords +from libecalc.presentation.yaml.yaml_types.emitters.yaml_venting_emitter import ( + YamlVentingEmitter, +) + + +class Installation(BaseComponent, EnergyComponent): + component_type: Literal[ComponentType.INSTALLATION] = ComponentType.INSTALLATION + + user_defined_category: Optional[InstallationUserDefinedCategoryType] = Field(default=None, validate_default=True) + hydrocarbon_export: dict[Period, Expression] + fuel_consumers: list[ + Annotated[ + Union[GeneratorSet, FuelConsumer, ConsumerSystem], + Field(discriminator="component_type"), + ] + ] = Field(default_factory=list) + venting_emitters: list[YamlVentingEmitter] = Field(default_factory=list) + + def is_fuel_consumer(self) -> bool: + return True + + def is_electricity_consumer(self) -> bool: + # Should maybe be True if power from shore? + return False + + def is_provider(self) -> bool: + return False + + def is_container(self) -> bool: + return True + + def get_component_process_type(self) -> ComponentType: + return self.component_type + + def get_name(self) -> str: + return self.name + + @property + def id(self) -> str: + return generate_id(self.name) + + _validate_installation_temporal_model = field_validator("hydrocarbon_export")(validate_temporal_model) + + _convert_expression_installation = field_validator("regularity", "hydrocarbon_export", mode="before")( + convert_expression + ) + + @field_validator("user_defined_category", mode="before") + def check_user_defined_category(cls, user_defined_category, info: ValidationInfo): + # Provide which value and context to make it easier for user to correct wrt mandatory changes. + if user_defined_category is not None: + if user_defined_category not in list(InstallationUserDefinedCategoryType): + name_context_str = "" + if (name := info.data.get("name")) is not None: + name_context_str = f"with the name {name}" + + raise ValueError( + f"CATEGORY: {user_defined_category} is not allowed for {cls.__name__} {name_context_str}. Valid categories are: {[str(installation_user_defined_category.value) for installation_user_defined_category in InstallationUserDefinedCategoryType]}" + ) + + return user_defined_category + + @model_validator(mode="after") + def check_fuel_consumers_or_venting_emitters_exist(self): + try: + if self.fuel_consumers or self.venting_emitters: + return self + except AttributeError: + raise ValueError( + f"Keywords are missing:\n It is required to specify at least one of the keywords " + f"{EcalcYamlKeywords.fuel_consumers}, {EcalcYamlKeywords.generator_sets} or {EcalcYamlKeywords.installation_venting_emitters} in the model.", + ) from None + + def get_graph(self) -> ComponentGraph: + graph = ComponentGraph() + graph.add_node(self) + for component in [*self.fuel_consumers, *self.venting_emitters]: + if hasattr(component, "get_graph"): + graph.add_subgraph(component.get_graph()) + else: + graph.add_node(component) + + graph.add_edge(self.id, component.id) + + return graph diff --git a/src/libecalc/domain/infrastructure/energy_components/pump/component_dto.py b/src/libecalc/domain/infrastructure/energy_components/pump/component_dto.py new file mode 100644 index 0000000000..d425ca53f0 --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/pump/component_dto.py @@ -0,0 +1,31 @@ +from typing import Literal + +from libecalc.application.energy.energy_component import EnergyComponent +from libecalc.common.component_type import ComponentType +from libecalc.common.consumption_type import ConsumptionType +from libecalc.common.time_utils import Period +from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseConsumer +from libecalc.dto.models.pump import PumpModel + + +class PumpComponent(BaseConsumer, EnergyComponent): + component_type: Literal[ComponentType.PUMP] = ComponentType.PUMP + energy_usage_model: dict[Period, PumpModel] + + def is_fuel_consumer(self) -> bool: + return self.consumes == ConsumptionType.FUEL + + def is_electricity_consumer(self) -> bool: + return self.consumes == ConsumptionType.ELECTRICITY + + def is_provider(self) -> bool: + return False + + def is_container(self) -> bool: + return False + + def get_component_process_type(self) -> ComponentType: + return self.component_type + + def get_name(self) -> str: + return self.name diff --git a/src/libecalc/domain/infrastructure/energy_components/utils.py b/src/libecalc/domain/infrastructure/energy_components/utils.py new file mode 100644 index 0000000000..02fedcab5a --- /dev/null +++ b/src/libecalc/domain/infrastructure/energy_components/utils.py @@ -0,0 +1,26 @@ +from datetime import datetime +from typing import Any, Union + +from libecalc.common.energy_usage_type import EnergyUsageType +from libecalc.common.time_utils import Period +from libecalc.dto.models import ConsumerFunction + + +def check_model_energy_usage_type(model_data: dict[Period, ConsumerFunction], energy_type: EnergyUsageType): + for model in model_data.values(): + if model.energy_usage_type != energy_type: + raise ValueError(f"Model does not consume {energy_type.value}") + return model_data + + +def _convert_keys_in_dictionary_from_str_to_periods(data: dict[Union[str, Period], Any]) -> dict[Period, Any]: + if all(isinstance(key, str) for key in data.keys()): + return { + Period( + start=datetime.strptime(period.split(";")[0], "%Y-%m-%d %H:%M:%S"), + end=datetime.strptime(period.split(";")[1], "%Y-%m-%d %H:%M:%S"), + ): value + for period, value in data.items() + } + else: + return data diff --git a/src/libecalc/presentation/exporter/infrastructure.py b/src/libecalc/presentation/exporter/infrastructure.py index 486be485d4..a20b126eb9 100644 --- a/src/libecalc/presentation/exporter/infrastructure.py +++ b/src/libecalc/presentation/exporter/infrastructure.py @@ -11,7 +11,7 @@ from libecalc.common.units import Unit from libecalc.common.utils.rates import TimeSeries, TimeSeriesFloat, TimeSeriesRate, TimeSeriesStreamDayRate from libecalc.core.result import GeneratorSetResult -from libecalc.domain.infrastructure.components import FuelConsumer, GeneratorSet +from libecalc.domain.infrastructure import FuelConsumer, GeneratorSet from libecalc.dto.utils.validators import convert_expression from libecalc.presentation.exporter.domain.exportable import ( Attribute, diff --git a/src/libecalc/presentation/json_result/mapper.py b/src/libecalc/presentation/json_result/mapper.py index b897ef6935..abd9507060 100644 --- a/src/libecalc/presentation/json_result/mapper.py +++ b/src/libecalc/presentation/json_result/mapper.py @@ -26,7 +26,7 @@ ) from libecalc.common.variables import ExpressionEvaluator from libecalc.core.result.emission import EmissionResult -from libecalc.domain.infrastructure import components +from libecalc.domain.infrastructure import Asset from libecalc.dto import CompressorSystemConsumerFunction from libecalc.expression import Expression from libecalc.presentation.json_result.aggregators import ( @@ -366,7 +366,7 @@ def get_asset_result(graph_result: GraphResult) -> libecalc.presentation.json_re asset_id = graph_result.graph.root asset = graph_result.graph.get_node(asset_id) - if not isinstance(asset, components.Asset): + if not isinstance(asset, Asset): raise ProgrammingError("Need an asset graph to get asset result") installation_results = _evaluate_installations( diff --git a/src/libecalc/presentation/yaml/mappers/component_mapper.py b/src/libecalc/presentation/yaml/mappers/component_mapper.py index 579bec6aa3..dfcdc02fae 100644 --- a/src/libecalc/presentation/yaml/mappers/component_mapper.py +++ b/src/libecalc/presentation/yaml/mappers/component_mapper.py @@ -8,14 +8,16 @@ from libecalc.common.energy_model_type import EnergyModelType from libecalc.common.logger import logger from libecalc.common.time_utils import Period, define_time_model_for_period -from libecalc.domain.infrastructure.components import ( +from libecalc.domain.infrastructure import ( Asset, - Consumer, - ElectricityConsumer, - FuelConsumer, - GeneratorSet, Installation, ) +from libecalc.domain.infrastructure.energy_components.common import Consumer +from libecalc.domain.infrastructure.energy_components.electricity_consumer.electricity_consumer import ( + ElectricityConsumer, +) +from libecalc.domain.infrastructure.energy_components.fuel_consumer.fuel_consumer import FuelConsumer +from libecalc.domain.infrastructure.energy_components.generator_set.generator_set_dto import GeneratorSet from libecalc.dto import ConsumerFunction, FuelType from libecalc.dto.utils.validators import convert_expression from libecalc.expression import Expression diff --git a/src/libecalc/presentation/yaml/yaml_types/components/system/yaml_consumer_system.py b/src/libecalc/presentation/yaml/yaml_types/components/system/yaml_consumer_system.py index 00ee2c21ad..7fdf6e9f5c 100644 --- a/src/libecalc/presentation/yaml/yaml_types/components/system/yaml_consumer_system.py +++ b/src/libecalc/presentation/yaml/yaml_types/components/system/yaml_consumer_system.py @@ -6,7 +6,8 @@ from libecalc.common.component_type import ComponentType from libecalc.common.consumption_type import ConsumptionType from libecalc.common.time_utils import Period, define_time_model_for_period -from libecalc.domain.infrastructure.components import ConsumerSystem, Crossover, SystemComponentConditions +from libecalc.domain.infrastructure.energy_components.base.component_dto import Crossover, SystemComponentConditions +from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system_dto import ConsumerSystem from libecalc.dto import FuelType from libecalc.expression import Expression from libecalc.presentation.yaml.domain.reference_service import ReferenceService diff --git a/src/libecalc/presentation/yaml/yaml_types/components/train/yaml_train.py b/src/libecalc/presentation/yaml/yaml_types/components/train/yaml_train.py index 8ec2c844e0..a3ddd025c4 100644 --- a/src/libecalc/presentation/yaml/yaml_types/components/train/yaml_train.py +++ b/src/libecalc/presentation/yaml/yaml_types/components/train/yaml_train.py @@ -7,7 +7,7 @@ from libecalc.common.component_type import ComponentType from libecalc.common.consumption_type import ConsumptionType from libecalc.common.time_utils import Period -from libecalc.domain.infrastructure.components import Stream, TrainComponent +from libecalc.domain.infrastructure.energy_components.common import Stream, TrainComponent from libecalc.expression import Expression from libecalc.presentation.yaml.domain.reference_service import ReferenceService from libecalc.presentation.yaml.yaml_types.components.yaml_base import YamlConsumerBase diff --git a/src/libecalc/presentation/yaml/yaml_types/components/yaml_compressor.py b/src/libecalc/presentation/yaml/yaml_types/components/yaml_compressor.py index 9fb265393a..b556ba7278 100644 --- a/src/libecalc/presentation/yaml/yaml_types/components/yaml_compressor.py +++ b/src/libecalc/presentation/yaml/yaml_types/components/yaml_compressor.py @@ -6,7 +6,7 @@ from libecalc.common.component_type import ComponentType from libecalc.common.consumption_type import ConsumptionType from libecalc.common.time_utils import Period, define_time_model_for_period -from libecalc.domain.infrastructure.components import CompressorComponent +from libecalc.domain.infrastructure.energy_components.compressor.component_dto import CompressorComponent from libecalc.dto import FuelType from libecalc.expression import Expression from libecalc.presentation.yaml.domain.reference_service import ReferenceService diff --git a/src/libecalc/presentation/yaml/yaml_types/components/yaml_pump.py b/src/libecalc/presentation/yaml/yaml_types/components/yaml_pump.py index 4c68ae6c16..4a3b44d3b0 100644 --- a/src/libecalc/presentation/yaml/yaml_types/components/yaml_pump.py +++ b/src/libecalc/presentation/yaml/yaml_types/components/yaml_pump.py @@ -6,7 +6,7 @@ from libecalc.common.component_type import ComponentType from libecalc.common.consumption_type import ConsumptionType from libecalc.common.time_utils import Period, define_time_model_for_period -from libecalc.domain.infrastructure.components import PumpComponent +from libecalc.domain.infrastructure.energy_components.pump.component_dto import PumpComponent from libecalc.dto import FuelType from libecalc.expression import Expression from libecalc.presentation.yaml.domain.reference_service import ReferenceService diff --git a/tests/libecalc/core/consumers/conftest.py b/tests/libecalc/core/consumers/conftest.py index 89b51d88ff..e2a1e2e190 100644 --- a/tests/libecalc/core/consumers/conftest.py +++ b/tests/libecalc/core/consumers/conftest.py @@ -4,7 +4,7 @@ import libecalc.common.energy_usage_type from libecalc import dto -from libecalc.domain.infrastructure import components +from libecalc.domain.infrastructure import ElectricityConsumer, FuelConsumer, GeneratorSet from libecalc.common.component_type import ComponentType from libecalc.common.time_utils import Period from libecalc.common.utils.rates import RateType @@ -31,7 +31,7 @@ def variables_map(methane_values): @pytest.fixture -def tabulated_fuel_consumer(fuel_gas) -> components.FuelConsumer: +def tabulated_fuel_consumer(fuel_gas) -> FuelConsumer: tabulated = dto.TabulatedConsumerFunction( model=dto.TabulatedData( headers=["RATE", "FUEL"], @@ -42,7 +42,7 @@ def tabulated_fuel_consumer(fuel_gas) -> components.FuelConsumer: variables=[dto.Variables(name="RATE", expression=Expression.setup_from_expression(value="RATE"))], energy_usage_type=libecalc.common.energy_usage_type.EnergyUsageType.FUEL, ) - return components.FuelConsumer( + return FuelConsumer( name="fuel_consumer", component_type=ComponentType.GENERIC, fuel=fuel_gas, @@ -53,8 +53,8 @@ def tabulated_fuel_consumer(fuel_gas) -> components.FuelConsumer: @pytest.fixture -def direct_el_consumer() -> components.ElectricityConsumer: - return components.ElectricityConsumer( +def direct_el_consumer() -> ElectricityConsumer: + return ElectricityConsumer( name="direct_consumer", component_type=ComponentType.GENERIC, user_defined_category={Period(datetime(1900, 1, 1)): "FIXED-PRODUCTION-LOAD"}, @@ -105,8 +105,8 @@ def generator_set_sampled_model_1000mw() -> dto.GeneratorSetSampled: @pytest.fixture -def genset_2mw_dto(fuel_dto, direct_el_consumer, generator_set_sampled_model_2mw) -> components.GeneratorSet: - return components.GeneratorSet( +def genset_2mw_dto(fuel_dto, direct_el_consumer, generator_set_sampled_model_2mw) -> GeneratorSet: + return GeneratorSet( name="genset", user_defined_category={Period(datetime(1900, 1, 1)): "TURBINE-GENERATOR"}, fuel={Period(datetime(1900, 1, 1)): fuel_dto}, @@ -119,10 +119,8 @@ def genset_2mw_dto(fuel_dto, direct_el_consumer, generator_set_sampled_model_2mw @pytest.fixture -def genset_1000mw_late_startup_dto( - fuel_dto, direct_el_consumer, generator_set_sampled_model_1000mw -) -> components.GeneratorSet: - return components.GeneratorSet( +def genset_1000mw_late_startup_dto(fuel_dto, direct_el_consumer, generator_set_sampled_model_1000mw) -> GeneratorSet: + return GeneratorSet( name="genset_late_startup", user_defined_category={Period(datetime(1900, 1, 1)): "TURBINE-GENERATOR"}, fuel={Period(datetime(1900, 1, 1)): fuel_dto}, diff --git a/tests/libecalc/core/consumers/test_consumer_utils.py b/tests/libecalc/core/consumers/test_consumer_utils.py index d37a9fdf66..7adeb2753a 100644 --- a/tests/libecalc/core/consumers/test_consumer_utils.py +++ b/tests/libecalc/core/consumers/test_consumer_utils.py @@ -3,11 +3,11 @@ import numpy as np from libecalc.common.string.string_utils import generate_id -from libecalc.domain.infrastructure.energy_components.consumer_system import ConsumerSystem +from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system import ConsumerSystem from libecalc.domain.infrastructure.energy_components.legacy_consumer.consumer_function.utils import ( apply_condition, ) -from libecalc.domain.infrastructure.components import Crossover +from libecalc.domain.infrastructure.energy_components.base.component_dto import Crossover @dataclass diff --git a/tests/libecalc/core/consumers/test_crossover.py b/tests/libecalc/core/consumers/test_crossover.py index 6cad9932f9..dfb8c6d75d 100644 --- a/tests/libecalc/core/consumers/test_crossover.py +++ b/tests/libecalc/core/consumers/test_crossover.py @@ -4,7 +4,7 @@ from libecalc.common.time_utils import Period from libecalc.common.units import Unit -from libecalc.domain.infrastructure.energy_components.consumer_system import ConsumerSystem +from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system import ConsumerSystem from libecalc.domain.stream_conditions import Pressure, Rate, StreamConditions diff --git a/tests/libecalc/core/models/test_fuel_model.py b/tests/libecalc/core/models/test_fuel_model.py index 676fd6ea9d..d6c551b184 100644 --- a/tests/libecalc/core/models/test_fuel_model.py +++ b/tests/libecalc/core/models/test_fuel_model.py @@ -8,7 +8,7 @@ from libecalc.common.units import Unit from libecalc.common.utils.rates import RateType, TimeSeriesRate from libecalc.common.variables import VariablesMap -from libecalc.domain.infrastructure.components import FuelModel +from libecalc.domain.infrastructure.energy_components.fuel_model.fuel_model import FuelModel from libecalc.expression import Expression diff --git a/tests/libecalc/dto/test_electricity_consumer.py b/tests/libecalc/dto/test_electricity_consumer.py index 08f42cf5ec..fa412dd876 100644 --- a/tests/libecalc/dto/test_electricity_consumer.py +++ b/tests/libecalc/dto/test_electricity_consumer.py @@ -4,7 +4,7 @@ from pydantic import ValidationError from libecalc import dto -from libecalc.domain.infrastructure import components +from libecalc.domain.infrastructure import ElectricityConsumer from libecalc.common.component_type import ComponentType from libecalc.common.energy_usage_type import EnergyUsageType from libecalc.common.time_utils import Period @@ -14,7 +14,7 @@ class TestElectricityConsumer: def test_invalid_energy_usage(self): with pytest.raises(ValidationError) as e: - components.ElectricityConsumer( + ElectricityConsumer( name="Test", component_type=ComponentType.GENERIC, user_defined_category={Period(datetime(1900, 1, 1)): "MISCELLANEOUS"}, @@ -29,7 +29,7 @@ def test_invalid_energy_usage(self): def test_valid_electricity_consumer(self): # Should not raise ValidationError - components.ElectricityConsumer( + ElectricityConsumer( name="Test", component_type=ComponentType.GENERIC, user_defined_category={Period(datetime(1900, 1, 1)): "MISCELLANEOUS"}, diff --git a/tests/libecalc/dto/test_fuel_consumer.py b/tests/libecalc/dto/test_fuel_consumer.py index 7cbbe909b4..2ce4e75720 100644 --- a/tests/libecalc/dto/test_fuel_consumer.py +++ b/tests/libecalc/dto/test_fuel_consumer.py @@ -6,7 +6,7 @@ import libecalc.dto.fuel_type import libecalc.dto.types from libecalc import dto -from libecalc.domain.infrastructure import components +from libecalc.domain.infrastructure import FuelConsumer, Installation from libecalc.common.component_type import ComponentType from libecalc.common.energy_usage_type import EnergyUsageType from libecalc.common.time_utils import Period @@ -40,7 +40,7 @@ def get_fuel(fuel_name: str, emission_name: str) -> dict[Period, libecalc.dto.fu } -def get_installation(installation_name: str, fuel_consumer: components.FuelConsumer) -> components.Installation: +def get_installation(installation_name: str, fuel_consumer: FuelConsumer) -> Installation: """ Generates an installation dto for use in testing @@ -51,7 +51,7 @@ def get_installation(installation_name: str, fuel_consumer: components.FuelConsu Returns: dto.Installation """ - return components.Installation( + return Installation( name=installation_name, regularity=regularity, hydrocarbon_export={Period(datetime(1900, 1, 1)): Expression.setup_from_expression("sim1;var1")}, @@ -64,7 +64,7 @@ def get_fuel_consumer( consumer_name: str, fuel_type: dict[Period, libecalc.dto.fuel_type.FuelType], category: dict[Period, libecalc.dto.types.ConsumerUserDefinedCategoryType], -) -> components.FuelConsumer: +) -> FuelConsumer: """ Generates a fuel consumer dto for use in testing @@ -76,7 +76,7 @@ def get_fuel_consumer( Returns: dto.FuelConsumer """ - return components.FuelConsumer( + return FuelConsumer( name=consumer_name, fuel=fuel_type, component_type=ComponentType.GENERIC, @@ -94,7 +94,7 @@ def get_fuel_consumer( class TestFuelConsumer: def test_missing_fuel(self): with pytest.raises(ValidationError) as exc_info: - components.FuelConsumer( + FuelConsumer( name="test", fuel={}, component_type=ComponentType.GENERIC, diff --git a/tests/libecalc/dto/test_generator_set.py b/tests/libecalc/dto/test_generator_set.py index e6a2db5b9f..2ae1c9ea08 100644 --- a/tests/libecalc/dto/test_generator_set.py +++ b/tests/libecalc/dto/test_generator_set.py @@ -5,7 +5,8 @@ import libecalc.dto.fuel_type from libecalc import dto -from libecalc.domain.infrastructure import components +from libecalc.domain.infrastructure import GeneratorSet, FuelConsumer +from libecalc.dto.models import GeneratorSetSampled from libecalc.common.component_type import ComponentType from libecalc.common.consumption_type import ConsumptionType from libecalc.common.energy_model_type import EnergyModelType @@ -40,11 +41,11 @@ def test_invalid_headers(self): class TestGeneratorSet: def test_valid(self): - generator_set_dto = components.GeneratorSet( + generator_set_dto = GeneratorSet( name="Test", user_defined_category={Period(datetime(1900, 1, 1)): "MISCELLANEOUS"}, generator_set_model={ - Period(datetime(1900, 1, 1)): components.GeneratorSetSampled( + Period(datetime(1900, 1, 1)): GeneratorSetSampled( headers=["FUEL", "POWER"], data=[[0, 0], [1, 2], [2, 4], [3, 6]], energy_usage_adjustment_constant=0.0, @@ -74,7 +75,7 @@ def test_genset_should_fail_with_fuel_consumer(self): name="fuel", emissions=[], ) - fuel_consumer = components.FuelConsumer( + fuel_consumer = FuelConsumer( name="test", fuel={Period(datetime(2000, 1, 1)): fuel}, consumes=ConsumptionType.FUEL, @@ -89,7 +90,7 @@ def test_genset_should_fail_with_fuel_consumer(self): user_defined_category={Period(datetime(2000, 1, 1)): ConsumerUserDefinedCategoryType.MISCELLANEOUS}, ) with pytest.raises(ValidationError): - components.GeneratorSet( + GeneratorSet( name="Test", user_defined_category={Period(datetime(1900, 1, 1)): ConsumerUserDefinedCategoryType.MISCELLANEOUS}, generator_set_model={}, @@ -105,7 +106,7 @@ def test_power_from_shore_wrong_category(self): # Check for CABLE_LOSS with pytest.raises(ValueError) as exc_info: - components.GeneratorSet( + GeneratorSet( name="Test", user_defined_category={Period(datetime(1900, 1, 1)): ConsumerUserDefinedCategoryType.BOILER}, generator_set_model={}, @@ -119,7 +120,7 @@ def test_power_from_shore_wrong_category(self): # Check for MAX_USAGE_FROM_SHORE with pytest.raises(ValueError) as exc_info: - components.GeneratorSet( + GeneratorSet( name="Test", user_defined_category={Period(datetime(1900, 1, 1)): ConsumerUserDefinedCategoryType.BOILER}, generator_set_model={}, @@ -134,7 +135,7 @@ def test_power_from_shore_wrong_category(self): ) with pytest.raises(ValueError) as exc_info: - components.GeneratorSet( + GeneratorSet( name="Test", user_defined_category={Period(datetime(1900, 1, 1)): ConsumerUserDefinedCategoryType.BOILER}, generator_set_model={}, diff --git a/tests/libecalc/input/mappers/test_model_mapper.py b/tests/libecalc/input/mappers/test_model_mapper.py index dd76bd9144..b1628c4aff 100644 --- a/tests/libecalc/input/mappers/test_model_mapper.py +++ b/tests/libecalc/input/mappers/test_model_mapper.py @@ -4,8 +4,7 @@ import pytest from pydantic import TypeAdapter -from libecalc import dto -from libecalc.domain.infrastructure import components +from libecalc.domain.infrastructure import Asset from libecalc.common.time_utils import Frequency, Period from libecalc.presentation.yaml.mappers.model import ModelMapper from libecalc.presentation.yaml.model import YamlModel @@ -295,7 +294,7 @@ def temporal_model_stream(temporal_model_source: str) -> ResourceStream: @pytest.fixture def temporal_yaml_model_factory(temporal_model_stream, resource_service_factory, configuration_service_factory): - def create_temporal_yaml_model(start: datetime, end: datetime) -> components.Asset: + def create_temporal_yaml_model(start: datetime, end: datetime) -> Asset: period = Period( start=start, end=end, diff --git a/tests/libecalc/output/flow_diagram/conftest.py b/tests/libecalc/output/flow_diagram/conftest.py index 96526535cd..7ff8e05167 100644 --- a/tests/libecalc/output/flow_diagram/conftest.py +++ b/tests/libecalc/output/flow_diagram/conftest.py @@ -5,7 +5,8 @@ import libecalc.common.energy_usage_type import libecalc.dto.fuel_type from libecalc import dto -from libecalc.domain.infrastructure import components +from libecalc.domain.infrastructure import FuelConsumer, Asset, Installation + from libecalc.common.component_type import ComponentType from libecalc.common.time_utils import Period from libecalc.expression import Expression @@ -59,8 +60,8 @@ def fuel_type_fd() -> libecalc.dto.fuel_type.FuelType: @pytest.fixture -def compressor_system_consumer_dto_fd(fuel_type_fd) -> components.FuelConsumer: - return components.FuelConsumer( +def compressor_system_consumer_dto_fd(fuel_type_fd) -> FuelConsumer: + return FuelConsumer( name="Compressor system 1", component_type=ComponentType.COMPRESSOR_SYSTEM, user_defined_category={Period(datetime.datetime(1900, 1, 1), datetime.datetime(2021, 1, 1)): "COMPRESSOR"}, @@ -143,8 +144,8 @@ def compressor_system_consumer_dto_fd(fuel_type_fd) -> components.FuelConsumer: @pytest.fixture -def compressor_consumer_dto_fd(fuel_type_fd) -> components.FuelConsumer: - return components.FuelConsumer( +def compressor_consumer_dto_fd(fuel_type_fd) -> FuelConsumer: + return FuelConsumer( name="Compressor 1", component_type=ComponentType.GENERIC, user_defined_category={Period(datetime.datetime(1900, 1, 1), datetime.datetime(2021, 1, 1)): "COMPRESSOR"}, @@ -163,13 +164,13 @@ def compressor_consumer_dto_fd(fuel_type_fd) -> components.FuelConsumer: @pytest.fixture def installation_with_dates_dto_fd( - compressor_system_consumer_dto_fd: components.FuelConsumer, - compressor_consumer_dto_fd: components.FuelConsumer, -) -> components.Asset: - return components.Asset( + compressor_system_consumer_dto_fd: FuelConsumer, + compressor_consumer_dto_fd: FuelConsumer, +) -> Asset: + return Asset( name="installation_with_dates", installations=[ - components.Installation( + Installation( name="Installation1", fuel_consumers=[compressor_system_consumer_dto_fd, compressor_consumer_dto_fd], regularity={ diff --git a/tests/libecalc/output/flow_diagram/test_ecalc_model_mapper.py b/tests/libecalc/output/flow_diagram/test_ecalc_model_mapper.py index 268832b2a6..4e3f630057 100644 --- a/tests/libecalc/output/flow_diagram/test_ecalc_model_mapper.py +++ b/tests/libecalc/output/flow_diagram/test_ecalc_model_mapper.py @@ -3,7 +3,7 @@ import pytest -from libecalc.domain.infrastructure import components +from libecalc.domain.infrastructure import Asset from libecalc.fixtures import YamlCase from libecalc.presentation.flow_diagram.energy_model_flow_diagram import ( EnergyModelFlowDiagram, @@ -32,9 +32,7 @@ def test_all_energy_usage_models(self, all_energy_usage_models_yaml: YamlCase, s assert first_subdiagram.end_date == datetime(2021, 1, 1) @pytest.mark.snapshot - def test_case_with_dates( - self, installation_with_dates_dto_fd: components.Asset, snapshot, energy_model_from_dto_factory - ): + def test_case_with_dates(self, installation_with_dates_dto_fd: Asset, snapshot, energy_model_from_dto_factory): model = energy_model_from_dto_factory(installation_with_dates_dto_fd) actual_fd = EnergyModelFlowDiagram( energy_model=model, model_period=list(installation_with_dates_dto_fd.installations[0].regularity.keys())[0] diff --git a/tests/libecalc/presentation/json_result/test_aggregators.py b/tests/libecalc/presentation/json_result/test_aggregators.py index 3fed743232..0c4709c7dc 100644 --- a/tests/libecalc/presentation/json_result/test_aggregators.py +++ b/tests/libecalc/presentation/json_result/test_aggregators.py @@ -6,7 +6,8 @@ import libecalc.dto.fuel_type import libecalc.dto.types from libecalc import dto -from libecalc.domain.infrastructure import components +from libecalc.common.component_type import ComponentType +from libecalc.domain.infrastructure import Asset, Installation, FuelConsumer from libecalc.application.energy_calculator import EnergyCalculator from libecalc.application.graph_result import GraphResult from libecalc.common.time_utils import Period, Periods @@ -25,7 +26,7 @@ def get_installation( name_inst: str, name_consumer: str, name_fuel: str, co2_factor: float, fuel_rate: float -) -> components.Installation: +) -> Installation: """Creates a simple installation object for use in asset setup Args: name_inst (str): Name of installation @@ -38,7 +39,7 @@ def get_installation( components.Installation """ - inst = components.Installation( + inst = Installation( name=name_inst, regularity={Period(datetime(1900, 1, 1)): Expression.setup_from_expression(1)}, hydrocarbon_export={Period(datetime(1900, 1, 1)): Expression.setup_from_expression(1)}, @@ -71,7 +72,7 @@ def fuel(name: str, co2_factor: float) -> libecalc.dto.fuel_type.FuelType: ) -def direct_fuel_consumer(name: str, name_fuel: str, co2_factor: float, fuel_rate: float) -> components.FuelConsumer: +def direct_fuel_consumer(name: str, name_fuel: str, co2_factor: float, fuel_rate: float) -> FuelConsumer: """Creates a simple direct fuel consumer object for use in installation setup Args: name (str): Name of direct fuel consumer @@ -83,9 +84,9 @@ def direct_fuel_consumer(name: str, name_fuel: str, co2_factor: float, fuel_rate components.FuelConsumer """ - return components.FuelConsumer( + return FuelConsumer( name=name, - component_type=components.ComponentType.GENERIC, + component_type=ComponentType.GENERIC, fuel={Period(datetime(2024, 1, 1)): fuel(name=name_fuel, co2_factor=co2_factor)}, regularity={Period(datetime(1900, 1, 1)): Expression.setup_from_expression(1)}, user_defined_category={ @@ -175,7 +176,7 @@ def test_aggregate_emissions_installations(self, energy_model_from_dto_factory): name_inst="INSTB", name_consumer="cons2", name_fuel="fuel2", co2_factor=10, fuel_rate=100 ) - asset = components.Asset( + asset = Asset( name="Main asset", installations=[inst_a, inst_b], ) From acfddb7b7e746e8852c842f4bfabe047f0bcc7a1 Mon Sep 17 00:00:00 2001 From: Frode Helgetun Krogh <70878501+frodehk@users.noreply.github.com> Date: Mon, 16 Dec 2024 13:28:53 +0100 Subject: [PATCH 2/2] chore: remove overload function --- .../consumer_system/consumer_system_dto.py | 10 +--------- 1 file changed, 1 insertion(+), 9 deletions(-) diff --git a/src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system_dto.py b/src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system_dto.py index bb1c850ed9..e52fd433e3 100644 --- a/src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system_dto.py +++ b/src/libecalc/domain/infrastructure/energy_components/consumer_system/consumer_system_dto.py @@ -1,5 +1,5 @@ from collections import defaultdict -from typing import Literal, Optional, Union, overload +from typing import Literal, Optional, Union from pydantic import Field @@ -228,14 +228,6 @@ def evaluate_stream_conditions( return dict(parsed_priorities) -@overload -def create_consumer(consumer: CompressorComponent, period: Period) -> Compressor: ... - - -@overload -def create_consumer(consumer: PumpComponent, period: Period) -> Pump: ... - - def create_consumer( consumer: Union[CompressorComponent, PumpComponent], period: Period,