From 062f3f4309d3e9f31daf49ff3928c2d90f7088b3 Mon Sep 17 00:00:00 2001 From: Frode Helgetun Krogh <70878501+frodehk@users.noreply.github.com> Date: Tue, 26 Nov 2024 13:38:23 +0100 Subject: [PATCH] chore: use builders for ltp tests (#711) * chore: replace dtos with builders for ltp tests ECALC-1813 --- src/libecalc/fixtures/__init__.py | 1 - .../cases/ltp_export/installation_setup.py | 541 ----- .../ltp_export/loading_storage_ltp_yaml.py | 63 - .../ltp_export/ltp_power_from_shore_yaml.py | 111 - src/libecalc/testing/yaml_builder.py | 5 +- .../presentation/exporter/conftest.py | 286 +++ .../presentation/exporter/test_ltp.py | 2164 ++++++++++++----- 7 files changed, 1820 insertions(+), 1351 deletions(-) delete mode 100644 src/libecalc/fixtures/cases/ltp_export/installation_setup.py delete mode 100644 src/libecalc/fixtures/cases/ltp_export/loading_storage_ltp_yaml.py delete mode 100644 src/libecalc/fixtures/cases/ltp_export/ltp_power_from_shore_yaml.py create mode 100644 tests/libecalc/presentation/exporter/conftest.py diff --git a/src/libecalc/fixtures/__init__.py b/src/libecalc/fixtures/__init__.py index bda9c379ec..6ca418c04a 100644 --- a/src/libecalc/fixtures/__init__.py +++ b/src/libecalc/fixtures/__init__.py @@ -6,7 +6,6 @@ ) from .cases.consumer_with_time_slots_models import * # noqa: F403 from .cases.ltp_export import ltp_export_yaml -from .cases.ltp_export.ltp_power_from_shore_yaml import ltp_pfs_yaml_factory from .compressor_process_simulations.compressor_process_simulations import * # noqa: F403 from .conftest import ( fuel_gas, diff --git a/src/libecalc/fixtures/cases/ltp_export/installation_setup.py b/src/libecalc/fixtures/cases/ltp_export/installation_setup.py deleted file mode 100644 index e36a3cbaf7..0000000000 --- a/src/libecalc/fixtures/cases/ltp_export/installation_setup.py +++ /dev/null @@ -1,541 +0,0 @@ -from datetime import datetime - -import numpy as np - -import libecalc.common.component_type -from libecalc.common.energy_usage_type import EnergyUsageType -from libecalc.common.time_utils import Period -from libecalc.common.units import Unit -from libecalc.common.utils.rates import RateType -from libecalc.dto import ( - CompressorConsumerFunction, - CompressorSampled, - DirectConsumerFunction, - ElectricityConsumer, - Emission, - FuelConsumer, - FuelType, - GeneratorSet, - GeneratorSetSampled, - Installation, -) -from libecalc.dto.types import ( - ConsumerUserDefinedCategoryType, - FuelTypeUserDefinedCategoryType, - InstallationUserDefinedCategoryType, -) -from libecalc.expression import Expression - -regularity_installation = 1.0 -regularity_consumer = 1.0 -power_usage_mw = 10 - -power_offshore_wind_mw = 1 -power_compressor_mw = 3 -compressor_rate = 3000000 -fuel_rate = 67000 -diesel_rate = 120000 -co2_factor = 1 -ch4_factor = 0.1 -nox_factor = 0.5 -nmvoc_factor = 0 - -date1 = datetime(2027, 1, 1) -date2 = datetime(2027, 4, 10) -date3 = datetime(2028, 1, 1) -date4 = datetime(2028, 4, 10) -date5 = datetime(2029, 1, 1) - -period1 = Period(date1, date2) -period2 = Period(date2, date3) -period3 = Period(date3, date4) -period4 = Period(date4, date5) -period5 = Period(date5) - -full_period = Period(datetime(1900, 1, 1)) -period_from_date1 = Period(date1) -period_from_date3 = Period(date3) - -regularity_temporal_installation = {full_period: Expression.setup_from_expression(regularity_installation)} -regularity_temporal_consumer = {full_period: Expression.setup_from_expression(regularity_consumer)} - -days_year1_first_half = period1.duration.days -days_year2_first_half = period3.duration.days - -days_year1_second_half = period2.duration.days -days_year2_second_half = period4.duration.days - - -def fuel_turbine() -> FuelType: - return FuelType( - name="fuel_gas", - emissions=[ - Emission( - name="co2", - factor=Expression.setup_from_expression(value=co2_factor), - ) - ], - user_defined_category=FuelTypeUserDefinedCategoryType.FUEL_GAS, - ) - - -def diesel_turbine() -> FuelType: - return FuelType( - name="diesel", - emissions=[ - Emission( - name="co2", - factor=Expression.setup_from_expression(value=co2_factor), - ) - ], - user_defined_category=FuelTypeUserDefinedCategoryType.DIESEL, - ) - - -def diesel_turbine_multi() -> FuelType: - return FuelType( - name="diesel", - emissions=[ - Emission( - name="co2", - factor=Expression.setup_from_expression(value=co2_factor), - ), - Emission( - name="ch4", - factor=Expression.setup_from_expression(value=ch4_factor), - ), - Emission( - name="nox", - factor=Expression.setup_from_expression(value=nox_factor), - ), - Emission( - name="nmvoc", - factor=Expression.setup_from_expression(value=nmvoc_factor), - ), - ], - user_defined_category=FuelTypeUserDefinedCategoryType.DIESEL, - ) - - -def generator_set_fuel() -> GeneratorSetSampled: - return GeneratorSetSampled( - headers=["POWER", "FUEL"], - data=[[0, 2.5, 5, power_usage_mw, 15, 20], [0, 30000, 45000, fuel_rate, 87000, 110000]], - energy_usage_adjustment_constant=0.0, - energy_usage_adjustment_factor=1.0, - ) - - -def generator_set_diesel() -> GeneratorSetSampled: - return GeneratorSetSampled( - headers=["POWER", "FUEL"], - data=[[0, power_usage_mw, 15, 20], [0, diesel_rate, 145000, 160000]], - energy_usage_adjustment_constant=0.0, - energy_usage_adjustment_factor=1.0, - ) - - -def fuel_dict() -> dict[Period, FuelType]: - return { - period1: diesel_turbine(), - period2: fuel_turbine(), - period3: diesel_turbine(), - period4: fuel_turbine(), - period5: diesel_turbine(), - } - - -def fuel_dict_multi() -> dict[Period, FuelType]: - return { - period1: diesel_turbine_multi(), - period2: fuel_turbine(), - period3: diesel_turbine_multi(), - period4: fuel_turbine(), - period5: diesel_turbine_multi(), - } - - -def generator_set_dict() -> dict[Period, GeneratorSetSampled]: - return { - period1: generator_set_diesel(), - period2: generator_set_fuel(), - period3: generator_set_diesel(), - period4: generator_set_fuel(), - period5: generator_set_diesel(), - } - - -def category_dict() -> dict[Period, ConsumerUserDefinedCategoryType]: - return { - period1: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, - period2: ConsumerUserDefinedCategoryType.POWER_FROM_SHORE, - period3: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, - period4: ConsumerUserDefinedCategoryType.POWER_FROM_SHORE, - period5: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, - } - - -def category_dict_coarse() -> dict[Period, ConsumerUserDefinedCategoryType]: - return { - period1: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, - period2: ConsumerUserDefinedCategoryType.POWER_FROM_SHORE, - period_from_date3: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, - } - - -def direct_consumer(power: float) -> DirectConsumerFunction: - return DirectConsumerFunction( - load=Expression.setup_from_expression(value=power), - energy_usage_type=EnergyUsageType.POWER, - ) - - -def offshore_wind() -> ElectricityConsumer: - return ElectricityConsumer( - name="direct_consumer", - component_type=libecalc.common.component_type.ComponentType.GENERIC, - user_defined_category={ - period1: ConsumerUserDefinedCategoryType.MISCELLANEOUS, - period2: ConsumerUserDefinedCategoryType.OFFSHORE_WIND, - period3: ConsumerUserDefinedCategoryType.MISCELLANEOUS, - period4: ConsumerUserDefinedCategoryType.OFFSHORE_WIND, - period5: ConsumerUserDefinedCategoryType.MISCELLANEOUS, - }, - energy_usage_model={ - period1: direct_consumer(power=0), - period2: direct_consumer(power=power_offshore_wind_mw), - period3: direct_consumer(power=0), - period4: direct_consumer(power=power_offshore_wind_mw), - period5: direct_consumer(power=0), - }, - regularity=regularity_temporal_consumer, - ) - - -def no_el_consumption() -> ElectricityConsumer: - return ElectricityConsumer( - name="no_el_consumption", - component_type=libecalc.common.component_type.ComponentType.GENERIC, - user_defined_category={full_period: ConsumerUserDefinedCategoryType.FIXED_PRODUCTION_LOAD}, - energy_usage_model={period_from_date1: direct_consumer(power=0)}, - regularity=regularity_temporal_consumer, - ) - - -def simple_direct_el_consumer(name: str = "direct_consumer") -> ElectricityConsumer: - return ElectricityConsumer( - name=name, - component_type=libecalc.common.component_type.ComponentType.GENERIC, - user_defined_category={full_period: ConsumerUserDefinedCategoryType.FIXED_PRODUCTION_LOAD}, - energy_usage_model={ - period_from_date1: DirectConsumerFunction( - load=Expression.setup_from_expression(value=power_usage_mw), - energy_usage_type=EnergyUsageType.POWER, - consumption_rate_type=RateType.STREAM_DAY, - ), - }, - regularity=regularity_temporal_consumer, - ) - - -def simple_direct_el_consumer_mobile() -> ElectricityConsumer: - return ElectricityConsumer( - name="direct_consumer_mobile", - component_type=libecalc.common.component_type.ComponentType.GENERIC, - user_defined_category={full_period: ConsumerUserDefinedCategoryType.FIXED_PRODUCTION_LOAD}, - energy_usage_model={ - period_from_date1: DirectConsumerFunction( - load=Expression.setup_from_expression(value=power_usage_mw), - energy_usage_type=EnergyUsageType.POWER, - consumption_rate_type=RateType.STREAM_DAY, - ), - }, - regularity=regularity_temporal_consumer, - ) - - -def compressor_sampled(): - return CompressorSampled( - energy_usage_type=EnergyUsageType.FUEL, - energy_usage_values=[0, 10000, 11000, 12000, 13000], - power_interpolation_values=[0.0, 1.0, 2.0, power_compressor_mw, 4.0], - rate_values=[0, 1000000, 2000000, compressor_rate, 4000000], - energy_usage_adjustment_constant=0.0, - energy_usage_adjustment_factor=1.0, - ) - - -def boiler_heater() -> FuelConsumer: - return FuelConsumer( - name="boiler", - component_type=libecalc.common.component_type.ComponentType.GENERIC, - fuel={period_from_date1: fuel_turbine()}, - user_defined_category={ - Period(date1, date4): ConsumerUserDefinedCategoryType.BOILER, - Period(date4): ConsumerUserDefinedCategoryType.HEATER, - }, - regularity=regularity_temporal_consumer, - energy_usage_model={ - full_period: DirectConsumerFunction( - fuel_rate=Expression.setup_from_expression(value=fuel_rate), - energy_usage_type=EnergyUsageType.FUEL, - ) - }, - ) - - -def compressor(name: str = "single_1d_compressor_sampled") -> FuelConsumer: - return FuelConsumer( - name=name, - component_type=libecalc.common.component_type.ComponentType.COMPRESSOR, - fuel={period_from_date1: fuel_turbine()}, - user_defined_category={ - period1: ConsumerUserDefinedCategoryType.MISCELLANEOUS, - period2: ConsumerUserDefinedCategoryType.GAS_DRIVEN_COMPRESSOR, - period3: ConsumerUserDefinedCategoryType.MISCELLANEOUS, - period4: ConsumerUserDefinedCategoryType.GAS_DRIVEN_COMPRESSOR, - period5: ConsumerUserDefinedCategoryType.MISCELLANEOUS, - }, - regularity=regularity_temporal_consumer, - energy_usage_model={ - full_period: CompressorConsumerFunction( - energy_usage_type=EnergyUsageType.FUEL, - model=compressor_sampled(), - rate_standard_m3_day=Expression.setup_from_expression(value=compressor_rate), - suction_pressure=Expression.setup_from_expression(value=200), - discharge_pressure=Expression.setup_from_expression(value=400), - ) - }, - ) - - -def generator_set_direct_consumer_temporal_model() -> GeneratorSet: - return GeneratorSet( - name="genset", - user_defined_category=category_dict_coarse(), - fuel=fuel_dict(), - generator_set_model=generator_set_dict(), - consumers=[simple_direct_el_consumer()], - regularity=regularity_temporal_consumer, - ) - - -def generator_set_offshore_wind_temporal_model() -> GeneratorSet: - return GeneratorSet( - name="genset", - user_defined_category={period_from_date1: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR}, - fuel={period_from_date1: fuel_turbine()}, - generator_set_model={period_from_date1: generator_set_fuel()}, - consumers=[offshore_wind()], - regularity=regularity_temporal_consumer, - ) - - -def generator_set_compressor_temporal_model(consumers: list[ElectricityConsumer], name: str = "genset") -> GeneratorSet: - return GeneratorSet( - name=name, - user_defined_category={period_from_date1: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR}, - fuel={period_from_date1: fuel_turbine()}, - generator_set_model={period_from_date1: generator_set_fuel()}, - consumers=consumers, - regularity=regularity_temporal_consumer, - ) - - -def generator_set_fixed_diesel() -> GeneratorSet: - return GeneratorSet( - name="genset_fixed", - user_defined_category=category_dict(), - fuel=fuel_dict_multi(), - generator_set_model=generator_set_dict(), - consumers=[simple_direct_el_consumer()], - regularity=regularity_temporal_consumer, - ) - - -def generator_set_mobile_diesel() -> GeneratorSet: - return GeneratorSet( - name="genset_mobile", - user_defined_category=category_dict(), - fuel=fuel_dict_multi(), - generator_set_model=generator_set_dict(), - consumers=[simple_direct_el_consumer_mobile()], - regularity=regularity_temporal_consumer, - ) - - -def expected_fuel_consumption() -> float: - n_days = np.sum(days_year2_second_half) - consumption = float(fuel_rate * n_days * regularity_consumer) - return consumption - - -def expected_diesel_consumption() -> float: - n_days = np.sum([days_year1_first_half, days_year2_first_half]) - consumption = float(diesel_rate * n_days * regularity_consumer) - return consumption - - -def expected_pfs_el_consumption() -> float: - n_days = np.sum(days_year1_second_half) - consumption_mw_per_day = power_usage_mw * n_days * regularity_consumer - consumption = float(Unit.MEGA_WATT_DAYS.to(Unit.GIGA_WATT_HOURS)(consumption_mw_per_day)) - return consumption - - -def expected_gas_turbine_el_generated() -> float: - n_days = np.sum([(days_year1_first_half + days_year2_first_half + days_year2_second_half)]) - consumption_mw_per_day = power_usage_mw * n_days * regularity_consumer - consumption = float(Unit.MEGA_WATT_DAYS.to(Unit.GIGA_WATT_HOURS)(consumption_mw_per_day)) - return consumption - - -def expected_co2_from_fuel() -> float: - emission_kg_per_day = float(fuel_rate * co2_factor) - emission_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_kg_per_day) - n_days = np.sum(days_year2_second_half) - emission_tons = float(emission_tons_per_day * n_days * regularity_consumer) - return emission_tons - - -def expected_co2_from_diesel() -> float: - emission_kg_per_day = float(diesel_rate * co2_factor) - emission_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_kg_per_day) - n_days = np.sum([days_year1_first_half, days_year2_first_half]) - emission_tons = float(emission_tons_per_day * n_days * regularity_consumer) - return emission_tons - - -def expected_ch4_from_diesel() -> float: - emission_kg_per_day = float(diesel_rate * ch4_factor) - emission_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_kg_per_day) - n_days = np.sum([days_year1_first_half, days_year2_first_half]) - emission_tons = float(emission_tons_per_day * n_days * regularity_consumer) - return emission_tons - - -def expected_nox_from_diesel() -> float: - emission_kg_per_day = float(diesel_rate * nox_factor) - emission_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_kg_per_day) - n_days = np.sum([days_year1_first_half, days_year2_first_half]) - emission_tons = float(emission_tons_per_day * n_days * regularity_consumer) - return emission_tons - - -def expected_nmvoc_from_diesel() -> float: - emission_kg_per_day = float(diesel_rate * nmvoc_factor) - emission_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_kg_per_day) - n_days = np.sum([days_year1_first_half, days_year2_first_half]) - emission_tons = float(emission_tons_per_day * n_days * regularity_consumer) - return emission_tons - - -def expected_offshore_wind_el_consumption() -> float: - n_days = np.sum([days_year1_second_half, days_year2_second_half]) - consumption_mw_per_day = power_offshore_wind_mw * n_days * regularity_consumer - consumption = -float(Unit.MEGA_WATT_DAYS.to(Unit.GIGA_WATT_HOURS)(consumption_mw_per_day)) - return consumption - - -# Why is this the only one without regularity? -def expected_gas_turbine_compressor_el_consumption() -> float: - n_days = np.sum([days_year1_second_half, days_year2_second_half]) - consumption_mw_per_day = power_compressor_mw * n_days - consumption = float(Unit.MEGA_WATT_DAYS.to(Unit.GIGA_WATT_HOURS)(consumption_mw_per_day)) - return consumption - - -def expected_boiler_fuel_consumption() -> float: - n_days = np.sum([days_year1_first_half, days_year1_second_half, days_year2_first_half]) - consumption = float(fuel_rate * n_days * regularity_consumer) - return consumption - - -def expected_heater_fuel_consumption() -> float: - n_days = np.sum(days_year2_second_half) - consumption = float(fuel_rate * n_days * regularity_consumer) - return consumption - - -def expected_co2_from_boiler() -> float: - emission_kg_per_day = float(fuel_rate * co2_factor) - emission_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_kg_per_day) - n_days = np.sum([days_year1_first_half, days_year1_second_half, days_year2_first_half]) - emission_tons = float(emission_tons_per_day * n_days * regularity_consumer) - return emission_tons - - -def expected_co2_from_heater() -> float: - emission_kg_per_day = float(fuel_rate * co2_factor) - emission_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_kg_per_day) - n_days = np.sum(days_year2_second_half) - emission_tons = float(emission_tons_per_day * n_days * regularity_consumer) - return emission_tons - - -def installation_direct_consumer_dto() -> Installation: - return Installation( - name="INSTALLATION_A", - regularity=regularity_temporal_installation, - hydrocarbon_export={full_period: Expression.setup_from_expression("sim1;var1")}, - fuel_consumers=[generator_set_direct_consumer_temporal_model()], - user_defined_category=InstallationUserDefinedCategoryType.FIXED, - ) - - -def installation_offshore_wind_dto() -> Installation: - return Installation( - name="INSTALLATION_A", - regularity=regularity_temporal_installation, - hydrocarbon_export={full_period: Expression.setup_from_expression("sim1;var1")}, - fuel_consumers=[generator_set_offshore_wind_temporal_model()], - user_defined_category=InstallationUserDefinedCategoryType.FIXED, - ) - - -def installation_compressor_dto( - el_consumers: list[ElectricityConsumer], - installation_name: str = "INSTALLATION_A", - genset_name: str = "genset", - compressor_name: str = "compressor", -) -> Installation: - return Installation( - name=installation_name, - regularity=regularity_temporal_installation, - hydrocarbon_export={full_period: Expression.setup_from_expression(0)}, - fuel_consumers=[ - generator_set_compressor_temporal_model(el_consumers, name=genset_name), - compressor(name=compressor_name), - ], - user_defined_category=InstallationUserDefinedCategoryType.FIXED, - ) - - -def installation_diesel_fixed_dto() -> Installation: - return Installation( - name="INSTALLATION_FIXED", - regularity=regularity_temporal_installation, - hydrocarbon_export={full_period: Expression.setup_from_expression("sim1;var1")}, - fuel_consumers=[generator_set_fixed_diesel()], - user_defined_category=InstallationUserDefinedCategoryType.FIXED, - ) - - -def installation_diesel_mobile_dto() -> Installation: - return Installation( - name="INSTALLATION_MOBILE", - regularity=regularity_temporal_installation, - hydrocarbon_export={full_period: Expression.setup_from_expression("sim1;var1")}, - fuel_consumers=[generator_set_mobile_diesel()], - user_defined_category=InstallationUserDefinedCategoryType.MOBILE, - ) - - -def installation_boiler_heater_dto() -> Installation: - return Installation( - name="INSTALLATION_A", - regularity=regularity_temporal_installation, - hydrocarbon_export={full_period: Expression.setup_from_expression("sim1;var1")}, - fuel_consumers=[boiler_heater()], - user_defined_category=InstallationUserDefinedCategoryType.FIXED, - ) diff --git a/src/libecalc/fixtures/cases/ltp_export/loading_storage_ltp_yaml.py b/src/libecalc/fixtures/cases/ltp_export/loading_storage_ltp_yaml.py deleted file mode 100644 index 27991eeca7..0000000000 --- a/src/libecalc/fixtures/cases/ltp_export/loading_storage_ltp_yaml.py +++ /dev/null @@ -1,63 +0,0 @@ -import yaml - -from libecalc.common.utils.rates import RateType -from libecalc.dto import Asset -from libecalc.presentation.yaml.parse_input import map_yaml_to_dto -from libecalc.presentation.yaml.yaml_models.pyyaml_yaml_model import PyYamlYamlModel - - -def ltp_oil_loaded_yaml_factory( - emission_factor: float, - rate_types: list[RateType], - fuel_rates: list[float], - emission_name: str, - regularity: float, - categories: list[str], - consumer_names: list[str], -) -> Asset: - input_text = f""" - FUEL_TYPES: - - NAME: fuel - EMISSIONS: - - NAME: {emission_name} - FACTOR: {emission_factor} - - INSTALLATIONS: - - NAME: minimal_installation - HCEXPORT: 0 - FUEL: fuel - CATEGORY: FIXED - REGULARITY: {regularity} - - FUELCONSUMERS: - {create_direct_consumers_yaml(categories, fuel_rates, rate_types, consumer_names)} - - """ - - create_direct_consumers_yaml(categories, fuel_rates, rate_types, consumer_names) - yaml_text = yaml.safe_load(input_text) - configuration = PyYamlYamlModel( - internal_datamodel=yaml_text, - name="test", - instantiated_through_read=True, - ) - yaml_model = map_yaml_to_dto(configuration=configuration, resources={}) - return yaml_model - - -def create_direct_consumers_yaml( - categories: list[str], fuel_rates: list[float], rate_types: list[RateType], consumer_names: list[str] -) -> str: - consumers = "" - for category, fuel_rate, rate_type, consumer_name in zip(categories, fuel_rates, rate_types, consumer_names): - consumer = f""" - - NAME: {consumer_name} - CATEGORY: {category} - FUEL: fuel - ENERGY_USAGE_MODEL: - TYPE: DIRECT - FUELRATE: {fuel_rate} - CONSUMPTION_RATE_TYPE: {rate_type.value} - """ - consumers = consumers + consumer - return consumers diff --git a/src/libecalc/fixtures/cases/ltp_export/ltp_power_from_shore_yaml.py b/src/libecalc/fixtures/cases/ltp_export/ltp_power_from_shore_yaml.py deleted file mode 100644 index 411d5988e1..0000000000 --- a/src/libecalc/fixtures/cases/ltp_export/ltp_power_from_shore_yaml.py +++ /dev/null @@ -1,111 +0,0 @@ -from io import StringIO -from pathlib import Path -from typing import Optional, cast - -import pytest - -from ecalc_cli.infrastructure.file_resource_service import FileResourceService -from libecalc.common.time_utils import Frequency -from libecalc.expression.expression import ExpressionType -from libecalc.fixtures.case_types import DTOCase -from libecalc.presentation.yaml.configuration_service import ConfigurationService -from libecalc.presentation.yaml.model import YamlModel -from libecalc.presentation.yaml.yaml_entities import ResourceStream -from libecalc.presentation.yaml.yaml_models.yaml_model import ReaderType, YamlConfiguration, YamlValidator - - -class OverridableStreamConfigurationService(ConfigurationService): - def __init__(self, stream: ResourceStream, overrides: Optional[dict] = None): - self._overrides = overrides - self._stream = stream - - def get_configuration(self) -> YamlValidator: - main_yaml_model = YamlConfiguration.Builder.get_yaml_reader(ReaderType.PYYAML).read( - main_yaml=self._stream, - enable_include=True, - ) - - if self._overrides is not None: - main_yaml_model._internal_datamodel.update(self._overrides) - return cast(YamlValidator, main_yaml_model) - - -@pytest.fixture -def ltp_pfs_yaml_factory(): - def _ltp_pfs_yaml_factory( - regularity: float, - cable_loss: ExpressionType, - max_usage_from_shore: ExpressionType, - load_direct_consumer: float, - path: Path, - ) -> DTOCase: - input_text = f""" - START: 2025-01-01 - END: 2031-01-01 - TIME_SERIES: - - NAME: CABLE_LOSS - TYPE: DEFAULT - FILE: data/sim/cable_loss.csv - - NAME: MAX_USAGE_FROM_SHORE - TYPE: DEFAULT - FILE: data/sim/max_usage_from_shore.csv - FACILITY_INPUTS: - - NAME: generator_energy_function - FILE: 'data/einput/genset_17MW.csv' - TYPE: ELECTRICITY2FUEL - - NAME: pfs_energy_function - FILE: 'data/einput/onshore_power.csv' - TYPE: ELECTRICITY2FUEL - FUEL_TYPES: - - NAME: fuel1 - EMISSIONS: - - NAME: co2 - FACTOR: 2 - - NAME: ch4 - FACTOR: 0.005 - - NAME: nmvoc - FACTOR: 0.002 - - NAME: nox - FACTOR: 0.001 - - INSTALLATIONS: - - NAME: minimal_installation - HCEXPORT: 0 - FUEL: fuel1 - CATEGORY: FIXED - REGULARITY: {regularity} - GENERATORSETS: - - NAME: generator1 - ELECTRICITY2FUEL: - 2025-01-01: generator_energy_function - 2027-01-01: pfs_energy_function - CATEGORY: - 2025-01-01: TURBINE-GENERATOR - 2027-01-01: POWER-FROM-SHORE - CABLE_LOSS: {cable_loss} - MAX_USAGE_FROM_SHORE: {max_usage_from_shore} - - CONSUMERS: # electrical energy consumers - - NAME: base_load - CATEGORY: BASE-LOAD - ENERGY_USAGE_MODEL: - TYPE: DIRECT - LOAD: {load_direct_consumer} - - - """ - - configuration_service = OverridableStreamConfigurationService( - stream=ResourceStream(name="ltp_export", stream=StringIO(input_text)) - ) - resource_service = FileResourceService(working_directory=path) - - model = YamlModel( - configuration_service=configuration_service, - resource_service=resource_service, - output_frequency=Frequency.YEAR, - ) - - return DTOCase(ecalc_model=model.dto, variables=model.variables) - - return _ltp_pfs_yaml_factory diff --git a/src/libecalc/testing/yaml_builder.py b/src/libecalc/testing/yaml_builder.py index 8341c32529..e070001157 100644 --- a/src/libecalc/testing/yaml_builder.py +++ b/src/libecalc/testing/yaml_builder.py @@ -10,6 +10,7 @@ FuelTypeUserDefinedCategoryType, InstallationUserDefinedCategoryType, ) + from libecalc.presentation.yaml.yaml_types import YamlBase from libecalc.presentation.yaml.yaml_types.components.legacy.energy_usage_model import ( YamlFuelEnergyUsageModel, @@ -140,7 +141,7 @@ def with_file(self, file: str): return self def with_test_data(self) -> Self: - self.name = "TimeSeriesDefault" + self.name = "DefaultTimeSeries" self.file = "DefaultTimeSeries.csv" return self @@ -281,7 +282,7 @@ def with_test_data(self) -> Self: self.name = "flare" self.category = ConsumerUserDefinedCategoryType.FLARE.value self.energy_usage_model = YamlEnergyUsageModelDirectBuilder().with_test_data().validate() - self.fuel = None + self.fuel = YamlFuelTypeBuilder().with_test_data().validate().name return self def with_name(self, name: str) -> Self: diff --git a/tests/libecalc/presentation/exporter/conftest.py b/tests/libecalc/presentation/exporter/conftest.py new file mode 100644 index 0000000000..6201b22a10 --- /dev/null +++ b/tests/libecalc/presentation/exporter/conftest.py @@ -0,0 +1,286 @@ +import pytest +from datetime import datetime +from typing import Optional, Union + +from libecalc.common.time_utils import Period +from libecalc.presentation.yaml.yaml_entities import MemoryResource +from libecalc.testing.yaml_builder import ( + YamlFuelTypeBuilder, + YamlFuelConsumerBuilder, + YamlEnergyUsageModelDirectBuilder, + YamlElectricityConsumerBuilder, +) + +from libecalc.dto.types import ( + FuelTypeUserDefinedCategoryType, + ConsumerUserDefinedCategoryType, +) + +from libecalc.presentation.yaml.yaml_types.components.legacy.energy_usage_model.yaml_energy_usage_model_direct import ( + ConsumptionRateType, +) + + +def memory_resource_factory(data: list[list[Union[float, int, str]]], headers: list[str]) -> MemoryResource: + return MemoryResource( + data=data, + headers=headers, + ) + + +# Memory resources +@pytest.fixture +def generator_electricity2fuel_17MW_resource(): + return memory_resource_factory( + data=[ + [0, 0.1, 10, 11, 12, 14, 15, 16, 17, 17.1, 18.5, 20, 20.5, 20.6, 24, 28, 30, 32, 34, 36, 38, 40, 41, 410], + [ + 0, + 75803.4, + 75803.4, + 80759.1, + 85714.8, + 95744, + 100728.8, + 105676.9, + 110598.4, + 136263.4, + 143260, + 151004.1, + 153736.5, + 154084.7, + 171429.6, + 191488, + 201457.5, + 211353.8, + 221196.9, + 231054, + 241049.3, + 251374.6, + 256839.4, + 2568394, + ], + ], # float and int with equal value should count as equal. + headers=[ + "POWER", + "FUEL", + ], + ) + + +@pytest.fixture +def onshore_power_electricity2fuel_resource(): + return memory_resource_factory( + data=[ + [0, 10, 20], + [0, 0, 0], + ], # float and int with equal value should count as equal. + headers=[ + "POWER", + "FUEL", + ], + ) + + +@pytest.fixture +def cable_loss_time_series_resource(): + return memory_resource_factory( + data=[ + [ + "01.01.2021", + "01.01.2022", + "01.01.2023", + "01.01.2024", + "01.01.2025", + "01.01.2026", + "01.01.2027", + "01.01.2028", + "01.01.2029", + "01.01.2030", + ], + [0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1], + ], # float and int with equal value should count as equal. + headers=[ + "DATE", + "CABLE_LOSS_FACTOR", + ], + ) + + +@pytest.fixture +def max_usage_from_shore_time_series_resource(): + return memory_resource_factory( + data=[ + [ + "01.01.2021", + "01.01.2022", + "01.01.2023", + "01.01.2024", + "01.01.2025", + "01.01.2026", + "01.01.2027", + "01.01.2028", + "01.01.2029", + "01.01.2030", + ], + [283, 283, 283, 283, 283, 250, 290, 283, 283, 283], + ], # float and int with equal value should count as equal. + headers=[ + "DATE", + "MAX_USAGE_FROM_SHORE", + ], + ) + + +@pytest.fixture +def compressor_sampled_fuel_driven_resource(): + def compressor(power_compressor_mw: Optional[float] = 3, compressor_rate: Optional[float] = 3000000): + return memory_resource_factory( + data=[ + [0.0, 1.0, 2.0, power_compressor_mw, 4.0], + [0, 10000, 11000, 12000, 13000], + [0, 1000000, 2000000, compressor_rate, 4000000], + ], # float and int with equal value should count as equal. + headers=["POWER", "FUEL", "RATE"], + ) + + return compressor + + +@pytest.fixture +def generator_diesel_power_to_fuel_resource(): + def generator(power_usage_mw: Optional[float] = 10, diesel_rate: Optional[float] = 120000): + return memory_resource_factory( + data=[ + [0, power_usage_mw, 15, 20], + [0, diesel_rate, 145000, 160000], + ], # float and int with equal value should count as equal. + headers=[ + "POWER", + "FUEL", + ], + ) + + return generator + + +@pytest.fixture +def generator_fuel_power_to_fuel_resource(): + def generator(power_usage_mw: Optional[float] = 10, fuel_rate: Optional[float] = 67000): + return memory_resource_factory( + data=[ + [0, 2.5, 5, power_usage_mw, 15, 20], + [0, 30000, 45000, fuel_rate, 87000, 110000], + ], # float and int with equal value should count as equal. + headers=[ + "POWER", + "FUEL", + ], + ) + + return generator + + +# Fixtures based on builders: +@pytest.fixture +def fuel_gas_factory(): + def fuel(names: Optional[list[str]] = None, factors: Optional[list[float]] = None): + if factors is None: + factors = [1] + if names is None: + names = ["co2"] + return ( + YamlFuelTypeBuilder() + .with_name("fuel") + .with_emission_names_and_factors(names=names, factors=factors) + .with_category(FuelTypeUserDefinedCategoryType.FUEL_GAS) + ).validate() + + return fuel + + +@pytest.fixture +def diesel_factory(): + def diesel_fuel(names=None, factors=None): + if factors is None: + factors = [1] + if names is None: + names = ["co2"] + return ( + YamlFuelTypeBuilder() + .with_name("diesel") + .with_emission_names_and_factors(names=names, factors=factors) + .with_category(FuelTypeUserDefinedCategoryType.DIESEL) + ).validate() + + return diesel_fuel + + +@pytest.fixture +def energy_usage_model_direct_factory(): + def energy_usage_model(rate: float): + return ( + YamlEnergyUsageModelDirectBuilder() + .with_fuel_rate(rate) + .with_consumption_rate_type(ConsumptionRateType.STREAM_DAY) + ).validate() + + return energy_usage_model + + +@pytest.fixture +def energy_usage_model_direct_load_factory(): + def energy_usage_model(load: float): + return ( + YamlEnergyUsageModelDirectBuilder() + .with_load(load) + .with_consumption_rate_type(ConsumptionRateType.STREAM_DAY) + ).validate() + + return energy_usage_model + + +@pytest.fixture +def fuel_consumer_direct_factory(energy_usage_model_direct_factory): + def fuel_consumer( + fuel_reference_name: str, + rate: float, + name: str = "fuel_consumer", + category: ConsumerUserDefinedCategoryType = ConsumerUserDefinedCategoryType.FLARE, + ): + return ( + YamlFuelConsumerBuilder() + .with_name(name) + .with_fuel(fuel_reference_name) + .with_category(category) + .with_energy_usage_model(energy_usage_model_direct_factory(rate)) + ).validate() + + return fuel_consumer + + +@pytest.fixture +def fuel_consumer_direct_load_factory(energy_usage_model_direct_load_factory): + def fuel_consumer(fuel_reference_name: str, load: float): + return ( + YamlFuelConsumerBuilder() + .with_name("fuel_consumer") + .with_fuel(fuel_reference_name) + .with_category(ConsumerUserDefinedCategoryType.BASE_LOAD) + .with_energy_usage_model(energy_usage_model_direct_load_factory(load)) + ).validate() + + return fuel_consumer + + +@pytest.fixture +def el_consumer_direct_base_load_factory(energy_usage_model_direct_load_factory): + def el_consumer(el_reference_name: str, load: float): + return ( + YamlElectricityConsumerBuilder() + .with_name(el_reference_name) + .with_category(ConsumerUserDefinedCategoryType.BASE_LOAD) + .with_energy_usage_model(energy_usage_model_direct_load_factory(load)) + ).validate() + + return el_consumer diff --git a/tests/libecalc/presentation/exporter/test_ltp.py b/tests/libecalc/presentation/exporter/test_ltp.py index 1ab3d4284e..f48434fa91 100644 --- a/tests/libecalc/presentation/exporter/test_ltp.py +++ b/tests/libecalc/presentation/exporter/test_ltp.py @@ -1,652 +1,1550 @@ +from copy import deepcopy from datetime import datetime +from io import StringIO from pathlib import Path from typing import Union +import numpy as np import pandas as pd import pytest -from libecalc import dto +from ecalc_cli.infrastructure.file_resource_service import FileResourceService from libecalc.application.energy_calculator import EnergyCalculator from libecalc.application.graph_result import GraphResult -from libecalc.common.time_utils import Period, calculate_delta_days +from libecalc.common.time_utils import Frequency, calculate_delta_days, Period, Periods from libecalc.common.units import Unit from libecalc.common.utils.rates import RateType from libecalc.common.variables import VariablesMap -from libecalc.fixtures.cases import ltp_export, venting_emitters -from libecalc.fixtures.cases.ltp_export.installation_setup import ( - expected_boiler_fuel_consumption, - expected_ch4_from_diesel, - expected_co2_from_boiler, - expected_co2_from_diesel, - expected_co2_from_fuel, - expected_co2_from_heater, - expected_diesel_consumption, - expected_fuel_consumption, - expected_gas_turbine_compressor_el_consumption, - expected_gas_turbine_el_generated, - expected_heater_fuel_consumption, - expected_nmvoc_from_diesel, - expected_nox_from_diesel, - expected_offshore_wind_el_consumption, - expected_pfs_el_consumption, - installation_boiler_heater_dto, - installation_compressor_dto, - installation_diesel_fixed_dto, - installation_diesel_mobile_dto, - installation_direct_consumer_dto, - installation_offshore_wind_dto, - no_el_consumption, - simple_direct_el_consumer, -) -from libecalc.fixtures.cases.ltp_export.loading_storage_ltp_yaml import ( - ltp_oil_loaded_yaml_factory, -) -from libecalc.fixtures.cases.ltp_export.utilities import ( - get_consumption, - get_sum_ltp_column, -) -from libecalc.fixtures.cases.venting_emitters.venting_emitter_yaml import ( - venting_emitter_yaml_factory, -) +from libecalc.dto.types import ConsumerUserDefinedCategoryType, InstallationUserDefinedCategoryType +from libecalc.presentation.exporter.configs.configs import LTPConfig +from libecalc.presentation.exporter.dto.dtos import FilteredResult, QueryResult +from libecalc.presentation.exporter.infrastructure import ExportableGraphResult from libecalc.presentation.json_result.mapper import get_asset_result from libecalc.presentation.json_result.result import EcalcModelResult -from libecalc.presentation.yaml.yaml_types.yaml_stream_conditions import ( - YamlEmissionRateUnits, +from libecalc.presentation.yaml.model import YamlModel +from libecalc.presentation.yaml.yaml_entities import MemoryResource, ResourceStream +from libecalc.presentation.yaml.yaml_models.pyyaml_yaml_model import PyYamlYamlModel +from libecalc.presentation.yaml.yaml_types.components.legacy.yaml_electricity_consumer import YamlElectricityConsumer +from libecalc.presentation.yaml.yaml_types.components.yaml_asset import YamlAsset +from libecalc.presentation.yaml.yaml_types.components.yaml_installation import YamlInstallation +from libecalc.presentation.yaml.yaml_types.fuel_type.yaml_fuel_type import YamlFuelType +from libecalc.presentation.yaml.yaml_types.yaml_stream_conditions import YamlEmissionRateUnits +from libecalc.testing.yaml_builder import ( + YamlAssetBuilder, + YamlVentingEmitterDirectTypeBuilder, + YamlInstallationBuilder, + YamlElectricity2fuelBuilder, + YamlTimeSeriesBuilder, + YamlGeneratorSetBuilder, + YamlEnergyUsageModelCompressorBuilder, + YamlCompressorTabularBuilder, + YamlFuelConsumerBuilder, + YamlEnergyUsageModelDirectBuilder, + YamlElectricityConsumerBuilder, ) -from libecalc.testing.dto_energy_model import DTOEnergyModel - -time_vector_installation = [ - datetime(2027, 1, 1), - datetime(2027, 4, 10), - datetime(2028, 1, 1), - datetime(2028, 4, 10), - datetime(2029, 1, 1), -] - -time_vector_yearly = pd.date_range(datetime(2027, 1, 1), datetime(2029, 1, 1), freq="YS").to_pydatetime().tolist() - - -def calculate_asset_result( - model: Union[dto.Installation, dto.Asset], - variables: VariablesMap, -): - energy_calculator = EnergyCalculator(energy_model=DTOEnergyModel(model), expression_evaluator=variables) - - consumer_results = energy_calculator.evaluate_energy_usage() - emission_results = energy_calculator.evaluate_emissions() - - graph = model.get_graph() - results_core = GraphResult( - graph=graph, - variables_map=variables, - consumer_results=consumer_results, - emission_results=emission_results, - ) - - results_dto = get_asset_result(results_core) - - return results_dto - - -def test_emissions_diesel_fixed_and_mobile(): - """Test reporting of CH4 from diesel in LTP.""" - installation_fixed = installation_diesel_fixed_dto() - installation_mobile = installation_diesel_mobile_dto() - - asset = dto.Asset( - name="multiple_installations_asset", - installations=[ - installation_fixed, - installation_mobile, - ], - ) - - variables = VariablesMap(time_vector=time_vector_installation, variables={"RATE": [1, 1, 1, 1]}) - - ltp_result = get_consumption(model=asset, variables=variables, periods=variables.get_periods()) - - co2_from_diesel_fixed = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="engineDieselCo2Mass") - co2_from_diesel_mobile = get_sum_ltp_column(ltp_result, installation_nr=1, ltp_column="engineNoCo2TaxDieselCo2Mass") - - nox_from_diesel_fixed = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="engineDieselNoxMass") - nox_from_diesel_mobile = get_sum_ltp_column(ltp_result, installation_nr=1, ltp_column="engineNoCo2TaxDieselNoxMass") - - nmvoc_from_diesel_fixed = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="engineDieselNmvocMass") - nmvoc_from_diesel_mobile = get_sum_ltp_column( - ltp_result, installation_nr=1, ltp_column="engineNoCo2TaxDieselNmvocMass" - ) - - ch4_from_diesel_fixed = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="engineDieselCh4Mass") - ch4_from_diesel_mobile = get_sum_ltp_column(ltp_result, installation_nr=1, ltp_column="engineNoCo2TaxDieselCh4Mass") - - assert co2_from_diesel_fixed == expected_co2_from_diesel() - assert co2_from_diesel_mobile == expected_co2_from_diesel() - assert nox_from_diesel_fixed == expected_nox_from_diesel() - assert nox_from_diesel_mobile == expected_nox_from_diesel() - assert nmvoc_from_diesel_fixed == expected_nmvoc_from_diesel() - assert nmvoc_from_diesel_mobile == expected_nmvoc_from_diesel() - assert ch4_from_diesel_fixed == expected_ch4_from_diesel() - assert ch4_from_diesel_mobile == expected_ch4_from_diesel() - - -def test_temporal_models_detailed(): - """Test various queries for LTP reporting. Purpose: ensure that variations in temporal models are captured. - - Detailed temporal models (variations within one year) for: - - Fuel type - - Generator set user defined category - - Generator set model - """ - variables = VariablesMap(time_vector=time_vector_installation, variables={"RATE": [1, 1, 1, 1]}) - - ltp_result = get_consumption( - model=installation_direct_consumer_dto(), variables=variables, periods=variables.get_periods() - ) - - turbine_fuel_consumption = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="turbineFuelGasConsumption") - engine_diesel_consumption = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="engineDieselConsumption") - - gas_turbine_el_generated = get_sum_ltp_column( - ltp_result, installation_nr=0, ltp_column="gasTurbineGeneratorConsumption" - ) - pfs_el_consumption = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="fromShoreConsumption") - co2_from_fuel = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="turbineFuelGasCo2Mass") - co2_from_diesel = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="engineDieselCo2Mass") - # FuelQuery: Check that turbine fuel consumption is included, - # even if the temporal model starts with diesel every year - assert turbine_fuel_consumption != 0 - - # FuelQuery: Check that turbine fuel consumption is correct - assert turbine_fuel_consumption == expected_fuel_consumption() - - # FuelQuery: Check that turbine fuel gas is not categorized as diesel, - # even if the temporal model starts with diesel every year - assert engine_diesel_consumption != expected_diesel_consumption() + expected_fuel_consumption() - - # FuelQuery: Check that diesel consumption is correct - assert engine_diesel_consumption == pytest.approx(expected_diesel_consumption(), 0.00001) - - # ElectricityGeneratedQuery: Check that turbine power generation is correct. - assert gas_turbine_el_generated == pytest.approx(expected_gas_turbine_el_generated(), 0.00001) - - # ElectricityGeneratedQuery: Check that power from shore el consumption is correct. - assert pfs_el_consumption == pytest.approx(expected_pfs_el_consumption(), 0.00001) - - # EmissionQuery. Check that co2 from fuel is correct. - assert co2_from_fuel == expected_co2_from_fuel() - - # EmissionQuery: Emissions. Check that co2 from diesel is correct. - assert co2_from_diesel == expected_co2_from_diesel() - - -def test_temporal_models_offshore_wind(): - """Test ElConsumerPowerConsumptionQuery for calculating offshore wind el-consumption, LTP. - - Detailed temporal models (variations within one year) for: - - El-consumer user defined category - - El-consumer energy usage model - """ - variables = VariablesMap(time_vector=time_vector_installation, variables={"RATE": [1, 1, 1, 1]}) - - ltp_result = get_consumption( - model=installation_offshore_wind_dto(), variables=variables, periods=variables.get_periods() - ) - - offshore_wind_el_consumption = get_sum_ltp_column( - ltp_result, installation_nr=0, ltp_column="offshoreWindConsumption" - ) - - # ElConsumerPowerConsumptionQuery: Check that offshore wind el-consumption is correct. - assert offshore_wind_el_consumption == expected_offshore_wind_el_consumption() - - -def test_temporal_models_compressor(): - """Test FuelConsumerPowerConsumptionQuery for calculating gas turbine compressor el-consumption, LTP. - - Detailed temporal models (variations within one year) for: - - Fuel consumer user defined category - """ - variables = VariablesMap(time_vector=time_vector_installation, variables={}) - - ltp_result = get_consumption( - model=installation_compressor_dto([no_el_consumption()]), variables=variables, periods=variables.get_periods() - ) - - gas_turbine_compressor_el_consumption = get_sum_ltp_column( - ltp_result, installation_nr=0, ltp_column="gasTurbineCompressorConsumption" - ) - - # FuelConsumerPowerConsumptionQuery. Check gas turbine compressor el consumption. - assert gas_turbine_compressor_el_consumption == expected_gas_turbine_compressor_el_consumption() - - -def test_boiler_heater_categories(): - variables = VariablesMap(time_vector=time_vector_installation, variables={}) - - ltp_result = get_consumption( - model=installation_boiler_heater_dto(), variables=variables, periods=variables.get_periods() - ) - - boiler_fuel_consumption = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="boilerFuelGasConsumption") - heater_fuel_consumption = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="heaterFuelGasConsumption") - co2_from_boiler = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="boilerFuelGasCo2Mass") - co2_from_heater = get_sum_ltp_column(ltp_result, installation_nr=0, ltp_column="heaterFuelGasCo2Mass") - - # FuelConsumerPowerConsumptionQuery. Check gas turbine compressor el consumption. - assert boiler_fuel_consumption == expected_boiler_fuel_consumption() - assert heater_fuel_consumption == expected_heater_fuel_consumption() - assert co2_from_boiler == expected_co2_from_boiler() - assert co2_from_heater == expected_co2_from_heater() - - -def test_venting_emitters(): - """Test venting emitters for LTP export. - - Verify correct behaviour if input rate is given in different units and rate types (sd and cd). - """ - time_vector = [ - datetime(2027, 1, 1), - datetime(2028, 1, 1), - ] - regularity = 0.2 - emission_rate = 10 - - variables = VariablesMap(time_vector=time_vector, variables={}) - - dto_sd_kg_per_day = venting_emitter_yaml_factory( - emission_rates=[emission_rate], - regularity=regularity, - units=[YamlEmissionRateUnits.KILO_PER_DAY], - emission_names=["ch4"], - rate_types=[RateType.STREAM_DAY], - names=["Venting emitter 1"], - path=Path(venting_emitters.__path__[0]), - ) - - dto_sd_tons_per_day = venting_emitter_yaml_factory( - emission_rates=[emission_rate], - regularity=regularity, - rate_types=[RateType.STREAM_DAY], - units=[YamlEmissionRateUnits.TONS_PER_DAY], - emission_names=["ch4"], - names=["Venting emitter 1"], - path=Path(venting_emitters.__path__[0]), - ) - - dto_cd_kg_per_day = venting_emitter_yaml_factory( - emission_rates=[emission_rate], - regularity=regularity, - rate_types=[RateType.CALENDAR_DAY], - units=[YamlEmissionRateUnits.KILO_PER_DAY], - emission_names=["ch4"], - names=["Venting emitter 1"], - path=Path(venting_emitters.__path__[0]), - ) - - ltp_result_input_sd_kg_per_day = get_consumption( - model=dto_sd_kg_per_day.ecalc_model, variables=variables, periods=variables.get_periods() - ) - - ltp_result_input_sd_tons_per_day = get_consumption( - model=dto_sd_tons_per_day.ecalc_model, variables=variables, periods=variables.get_periods() - ) - - ltp_result_input_cd_kg_per_day = get_consumption( - model=dto_cd_kg_per_day.ecalc_model, variables=variables, periods=variables.get_periods() - ) - - emission_input_sd_kg_per_day = get_sum_ltp_column( - ltp_result_input_sd_kg_per_day, installation_nr=0, ltp_column="storageCh4Mass" - ) - emission_input_sd_tons_per_day = get_sum_ltp_column( - ltp_result_input_sd_tons_per_day, installation_nr=0, ltp_column="storageCh4Mass" - ) - emission_input_cd_kg_per_day = get_sum_ltp_column( - ltp_result_input_cd_kg_per_day, installation_nr=0, ltp_column="storageCh4Mass" - ) - - # Verify correct emissions when input is kg per day. Output should be in tons per day - hence dividing by 1000 - assert emission_input_sd_kg_per_day == (emission_rate / 1000) * 365 * regularity - - # Verify correct emissions when input is tons per day. - assert emission_input_sd_tons_per_day == emission_rate * 365 * regularity - - # Verify that input calendar day vs input stream day is linked correctly through regularity - assert emission_input_cd_kg_per_day == emission_input_sd_kg_per_day / regularity - - # Verify that results is independent of regularity, when input rate is in calendar days - assert emission_input_cd_kg_per_day == (emission_rate / 1000) * 365 - - -def test_only_venting_emitters_no_fuelconsumers(): - """ - Test that it is possible with only venting emitters, without fuelconsumers. - """ - time_vector = [ - datetime(2027, 1, 1), - datetime(2028, 1, 1), - ] - regularity = 0.2 - emission_rate = 10 - - variables = VariablesMap(time_vector=time_vector, variables={}) - - # Installation with only venting emitters: - dto_case_emitters = venting_emitter_yaml_factory( - emission_rates=[emission_rate], - regularity=regularity, - units=[YamlEmissionRateUnits.KILO_PER_DAY], - emission_names=["ch4"], - rate_types=[RateType.STREAM_DAY], - include_emitters=True, - include_fuel_consumers=False, - names=["Venting emitter 1"], - installation_name="Venting emitter installation", - path=Path(venting_emitters.__path__[0]), - ) - - venting_emitter_results = get_consumption( - model=dto_case_emitters.ecalc_model, variables=variables, periods=variables.get_periods() - ) - - # Verify that eCalc is not failing in get_asset_result with only venting emitters - - # when installation result is empty, i.e. with no genset and fuel consumers: - assert isinstance( - calculate_asset_result(model=dto_case_emitters.ecalc_model, variables=variables), EcalcModelResult - ) - - # Verify correct emissions: - emissions_ch4 = get_sum_ltp_column(venting_emitter_results, installation_nr=0, ltp_column="storageCh4Mass") - assert emissions_ch4 == (emission_rate / 1000) * 365 * regularity - - # Installation with only fuel consumers: - dto_case_fuel = venting_emitter_yaml_factory( - emission_rates=[emission_rate], - regularity=regularity, - units=[YamlEmissionRateUnits.KILO_PER_DAY], - emission_names=["ch4"], - rate_types=[RateType.STREAM_DAY], - include_emitters=False, - include_fuel_consumers=True, - names=["Venting emitter 1"], - installation_name="Fuel consumer installation", - path=Path(venting_emitters.__path__[0]), - ) - - asset_multi_installations = dto.Asset( - name="Multi installations", - installations=[dto_case_emitters.ecalc_model.installations[0], dto_case_fuel.ecalc_model.installations[0]], - ) - - # Verify that eCalc is not failing in get_asset_result, with only venting emitters - - # when installation result is empty for one installation, i.e. with no genset and fuel consumers. - # Include asset with two installations, one with only emitters and one with only fuel consumers - - # ensure that get_asset_result returns a result: - - assert isinstance(calculate_asset_result(model=asset_multi_installations, variables=variables), EcalcModelResult) - - asset_ltp_result = get_consumption( - model=asset_multi_installations, variables=variables, periods=variables.get_periods() - ) - # Check that the results are the same: For the case with only one installation (only venting emitters), - # compared to the multi-installation case with two installations. The fuel-consumer installation should - # give no CH4-contribution (only CO2) - emissions_ch4_asset = get_sum_ltp_column(asset_ltp_result, installation_nr=0, ltp_column="storageCh4Mass") - assert emissions_ch4 == emissions_ch4_asset - - -def test_total_oil_loaded_old_method(): - """Test total oil loaded/stored for LTP export. Using original method where direct/venting emitters are - modelled as FUELSCONSUMERS using DIRECT. - - Verify correct volume when model includes emissions related to both storage and loading of oil, - and when model includes only loading. - """ - time_vector = [ - datetime(2027, 1, 1), - datetime(2028, 1, 1), - ] - variables = VariablesMap(time_vector=time_vector, variables={}) - - regularity = 0.6 - emission_factor = 2 - fuel_rate = 100 - - # Create model with both loading and storage - asset_loading_storage = ltp_oil_loaded_yaml_factory( - emission_factor=emission_factor, - rate_types=[RateType.STREAM_DAY, RateType.STREAM_DAY], - fuel_rates=[fuel_rate, fuel_rate], - emission_name="ch4", - regularity=regularity, - categories=["LOADING", "STORAGE"], - consumer_names=["loading", "storage"], - ) - - # Create model with only loading, not storage - asset_loading_only = ltp_oil_loaded_yaml_factory( - emission_factor=emission_factor, - rate_types=[RateType.STREAM_DAY], - fuel_rates=[fuel_rate], - emission_name="ch4", - regularity=regularity, - categories=["LOADING"], - consumer_names=["loading"], - ) - - ltp_result_loading_storage = get_consumption( - model=asset_loading_storage, variables=variables, periods=variables.get_periods() - ) - ltp_result_loading_only = get_consumption( - model=asset_loading_only, variables=variables, periods=variables.get_periods() - ) - - loaded_and_stored_oil_loading_and_storage = get_sum_ltp_column( - ltp_result_loading_storage, installation_nr=0, ltp_column="loadedAndStoredOil" - ) - loaded_and_stored_oil_loading_only = get_sum_ltp_column( - ltp_result_loading_only, installation_nr=0, ltp_column="loadedAndStoredOil" - ) - - # Verify output for total oil loaded/stored, if only loading is specified. - assert loaded_and_stored_oil_loading_only is not None - - # Verify correct volume for oil loaded/stored - assert loaded_and_stored_oil_loading_and_storage == fuel_rate * 365 * regularity - - # Verify that total oil loaded/stored is the same if only loading is specified, - # compared to a model with both loading and storage. - assert loaded_and_stored_oil_loading_and_storage == loaded_and_stored_oil_loading_only - - -def test_electrical_and_mechanical_power_installation(): - """Check that new total power includes the sum of electrical- and mechanical power at installation level""" - variables = VariablesMap(time_vector=time_vector_installation, variables={}) - asset = dto.Asset( - name="Asset 1", - installations=[ - installation_compressor_dto([simple_direct_el_consumer()]), - ], - ) - - asset_result = calculate_asset_result(model=asset, variables=variables) - power_fuel_driven_compressor = asset_result.get_component_by_name("compressor").power_cumulative.values[-1] - power_generator_set = asset_result.get_component_by_name("genset").power_cumulative.values[-1] - - # Extract cumulative electrical-, mechanical- and total power. - power_electrical_installation = asset_result.get_component_by_name( - "INSTALLATION_A" - ).power_electrical_cumulative.values[-1] - - power_mechanical_installation = asset_result.get_component_by_name( - "INSTALLATION_A" - ).power_mechanical_cumulative.values[-1] - - power_total_installation = asset_result.get_component_by_name("INSTALLATION_A").power_cumulative.values[-1] - - # Verify that total power is correct - assert power_total_installation == power_electrical_installation + power_mechanical_installation - - # Verify that electrical power equals genset power, and mechanical power equals power from gas driven compressor: - assert power_generator_set == power_electrical_installation - assert power_fuel_driven_compressor == power_mechanical_installation - - -def test_electrical_and_mechanical_power_asset(): - """Check that new total power includes the sum of electrical- and mechanical power at installation level""" - variables = VariablesMap(time_vector=time_vector_installation, variables={}) - installation_name_1 = "INSTALLATION_1" - installation_name_2 = "INSTALLATION_2" - - asset = dto.Asset( - name="Asset 1", - installations=[ - installation_compressor_dto( - [simple_direct_el_consumer(name="direct_el_consumer 1")], - installation_name=installation_name_1, - genset_name="generator 1", - compressor_name="gas driven compressor 1", +from libecalc.presentation.yaml.yaml_types.components.legacy.energy_usage_model.yaml_energy_usage_model_direct import ( + ConsumptionRateType, +) +from tests.conftest import DirectResourceService, OverridableStreamConfigurationService, yaml_model_factory + + +class LtpTestHelper: + def __init__(self): + # Constants + self.power_usage_mw = 10 + self.power_offshore_wind_mw = 1 + self.power_compressor_mw = 3 + self.fuel_rate = 67000 + self.diesel_rate = 120000 + self.load_consumer = 10 + self.compressor_rate = 3000000 + self.regularity_installation = 1.0 + self.co2_factor = 1 + self.ch4_factor = 0.1 + self.nox_factor = 0.5 + self.nmvoc_factor = 0 + + # Dates + self.date1 = datetime(2027, 1, 1) + self.date2 = datetime(2027, 4, 10) + self.date3 = datetime(2028, 1, 1) + self.date4 = datetime(2028, 4, 10) + self.date5 = datetime(2029, 1, 1) + self.time_vector_installation = [self.date1, self.date2, self.date3, self.date4, self.date5] + + # Periods + self.period1 = Period(self.date1, self.date2) + self.period2 = Period(self.date2, self.date3) + self.period3 = Period(self.date3, self.date4) + self.period4 = Period(self.date4, self.date5) + self.period5 = Period(self.date5) + self.period_from_date1 = Period(self.date1) + self.period_from_date3 = Period(self.date3) + self.full_period = Period(datetime(1900, 1, 1)) + + # Days + self.days_year1_first_half = self.period1.duration.days + self.days_year2_first_half = self.period3.duration.days + self.days_year1_second_half = self.period2.duration.days + self.days_year2_second_half = self.period4.duration.days + + def get_yaml_model( + self, + request, + asset: YamlAsset, + resources: dict[str, MemoryResource], + frequency: Frequency = Frequency.NONE, + ) -> YamlModel: + yaml_model_factory = request.getfixturevalue("yaml_model_factory") + asset_dict = asset.model_dump( + serialize_as_any=True, + mode="json", + exclude_unset=True, + by_alias=True, + ) + + yaml_string = PyYamlYamlModel.dump_yaml(yaml_dict=asset_dict) + stream = ResourceStream(name="", stream=StringIO(yaml_string)) + + return yaml_model_factory(resource_stream=stream, resources=resources, frequency=frequency) + + def get_consumption( + self, + model: Union[YamlInstallation, YamlAsset, YamlModel], + variables: VariablesMap, + frequency: Frequency, + periods: Periods, + ) -> FilteredResult: + energy_calculator = EnergyCalculator(energy_model=model, expression_evaluator=variables) + precision = 6 + + consumer_results = energy_calculator.evaluate_energy_usage() + emission_results = energy_calculator.evaluate_emissions() + + graph_result = GraphResult( + graph=model.get_graph(), + variables_map=variables, + consumer_results=consumer_results, + emission_results=emission_results, + ) + + ltp_filter = LTPConfig.filter(frequency=frequency) + ltp_result = ltp_filter.filter(ExportableGraphResult(graph_result), periods) + + return ltp_result + + def get_sum_ltp_column(self, ltp_result: FilteredResult, installation_nr, ltp_column: str) -> float: + installation_query_results = ltp_result.query_results[installation_nr].query_results + column = [column for column in installation_query_results if column.id == ltp_column][0] + + ltp_sum = sum(float(v) for (k, v) in column.values.items()) + return ltp_sum + + def get_ltp_column(self, ltp_result: FilteredResult, installation_nr, ltp_column: str) -> QueryResult: + installation_query_results = ltp_result.query_results[installation_nr].query_results + column = [column for column in installation_query_results if column.id == ltp_column][0] + + return column + + def get_ltp_result(self, model, variables, frequency=Frequency.YEAR): + return self.get_consumption( + model=model, variables=variables, periods=variables.get_periods(), frequency=frequency + ) + + def create_variables_map(self, time_vector, rate_values=None): + variables = {"RATE": rate_values} if rate_values else {} + return VariablesMap(time_vector=time_vector, variables=variables) + + def calculate_asset_result(self, model: YamlModel, variables: VariablesMap): + model = model + graph = model.get_graph() + energy_calculator = EnergyCalculator(energy_model=model, expression_evaluator=variables) + + consumer_results = energy_calculator.evaluate_energy_usage() + emission_results = energy_calculator.evaluate_emissions() + + results_core = GraphResult( + graph=graph, + variables_map=variables, + consumer_results=consumer_results, + emission_results=emission_results, + ) + + results_dto = get_asset_result(results_core) + + return results_dto + + def assert_emissions(self, ltp_result, installation_nr, ltp_column, expected_value): + actual_value = self.get_sum_ltp_column(ltp_result, installation_nr, ltp_column) + assert actual_value == pytest.approx(expected_value, 0.00001) + + def assert_consumption(self, ltp_result, installation_nr, ltp_column, expected_value): + actual_value = self.get_sum_ltp_column(ltp_result, installation_nr, ltp_column) + assert actual_value == pytest.approx(expected_value, 0.00001) + + def emission_calculate(self, rate: float, factor: float, days: list[int], regularity: float = None): + if regularity is None: + regularity = self.regularity_installation + emission_kg_per_day = float(rate * factor) + emission_tons_per_day = Unit.KILO_PER_DAY.to(Unit.TONS_PER_DAY)(emission_kg_per_day) + n_days = np.sum(days) + emission_tons = float(emission_tons_per_day * n_days * regularity) + return emission_tons + + def consumption_calculate(self, rate: float, days: list[int], regularity: float = None): + if regularity is None: + regularity = self.regularity_installation + n_days = np.sum(days) + return float(rate * n_days * regularity) + + def el_consumption_calculate(self, power: float, days: list[int], regularity: float = None): + if regularity is None: + regularity = self.regularity_installation + n_days = np.sum(days) + consumption_mw_per_day = power * n_days * regularity + consumption = float(Unit.MEGA_WATT_DAYS.to(Unit.GIGA_WATT_HOURS)(consumption_mw_per_day)) + return consumption + + def fuel_consumer_compressor(self, fuel: str, name: str = "compressor"): + return ( + YamlFuelConsumerBuilder() + .with_name(name) + .with_fuel({self.period_from_date1.start: fuel}) + .with_energy_usage_model(self.compressor_energy_usage_model) + .with_category( + self.temporal_dict( + reference1=ConsumerUserDefinedCategoryType.MISCELLANEOUS, + reference2=ConsumerUserDefinedCategoryType.GAS_DRIVEN_COMPRESSOR, + ) + ) + ).validate() + + def fuel_multi_temporal(self, fuel1: YamlFuelType, fuel2: YamlFuelType): + return { + self.period1.start: fuel1.name, + self.period2.start: fuel2.name, + self.period3.start: fuel1.name, + self.period4.start: fuel2.name, + self.period5.start: fuel1.name, + } + + def offshore_wind_consumer(self, request, power_mw: float = 1): + energy_usage_model_direct_load = request.getfixturevalue("energy_usage_model_direct_load_factory") + return ( + YamlElectricityConsumerBuilder() + .with_name("offshore_wind_consumer") + .with_category( + { + self.period1.start: ConsumerUserDefinedCategoryType.MISCELLANEOUS, + self.period2.start: ConsumerUserDefinedCategoryType.OFFSHORE_WIND, + self.period3.start: ConsumerUserDefinedCategoryType.MISCELLANEOUS, + self.period4.start: ConsumerUserDefinedCategoryType.OFFSHORE_WIND, + self.period5.start: ConsumerUserDefinedCategoryType.MISCELLANEOUS, + } + ) + .with_energy_usage_model( + { + self.period1.start: energy_usage_model_direct_load(load=0), + self.period2.start: energy_usage_model_direct_load(load=power_mw), + self.period3.start: energy_usage_model_direct_load(load=0), + self.period4.start: energy_usage_model_direct_load(load=power_mw), + self.period5.start: energy_usage_model_direct_load(load=0), + } + ) + ).validate() + + def generator_set( + self, + request, + fuel: str = None, + el_consumer: YamlElectricityConsumer = None, + el2fuel: Union[str, dict[datetime, str]] = None, + category: dict[datetime, ConsumerUserDefinedCategoryType] = None, + date: datetime = None, + name: str = "generator_set", + ): + if date is None: + date = self.period_from_date1.start + + if el_consumer is None: + direct_load = request.getfixturevalue("el_consumer_direct_base_load_factory") + el_consumer = direct_load(el_reference_name="base_load", load=self.load_consumer) + + if fuel is None: + fuel_gas = request.getfixturevalue("fuel_gas_factory") + fuel = fuel_gas(["co2"], [self.co2_factor]).name + + if el2fuel is None: + el2fuel = {date: self.generator_fuel_energy_function.name} + + if category is None: + category = {date: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR} + + return ( + YamlGeneratorSetBuilder() + .with_name(name) + .with_fuel(fuel) + .with_category(category) + .with_electricity2fuel(el2fuel) + .with_consumers([el_consumer]) + ).validate() + + def temporal_dict(self, reference1: str, reference2: str): + return { + self.period1.start: reference1, + self.period2.start: reference2, + self.period3.start: reference1, + self.period4.start: reference2, + self.period5.start: reference1, + } + + def category_dict(self) -> dict[datetime, ConsumerUserDefinedCategoryType]: + return { + self.period1.start: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, + self.period2.start: ConsumerUserDefinedCategoryType.POWER_FROM_SHORE, + self.period3.start: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, + self.period4.start: ConsumerUserDefinedCategoryType.POWER_FROM_SHORE, + self.period5.start: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, + } + + def category_dict_coarse(self) -> dict[datetime, ConsumerUserDefinedCategoryType]: + return { + self.period1.start: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, + self.period2.start: ConsumerUserDefinedCategoryType.POWER_FROM_SHORE, + self.period_from_date3.start: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, + } + + @property + def generator_fuel_energy_function(self): + return ( + YamlElectricity2fuelBuilder() + .with_name("generator_fuel_energy_function") + .with_file("generator_fuel_energy_function") + ).validate() + + @property + def generator_diesel_energy_function(self): + return ( + YamlElectricity2fuelBuilder() + .with_name("generator_diesel_energy_function") + .with_file("generator_diesel_energy_function") + ).validate() + + @property + def compressor_energy_function(self): + return ( + YamlCompressorTabularBuilder() + .with_name("compressor_energy_function") + .with_file("compressor_energy_function") + ).validate() + + @property + def power_from_shore_energy_function(self): + return ( + YamlElectricity2fuelBuilder().with_name("pfs_energy_function").with_file("pfs_energy_function") + ).validate() + + @property + def compressor_energy_usage_model(self): + return ( + YamlEnergyUsageModelCompressorBuilder() + .with_rate(self.compressor_rate) + .with_suction_pressure(200) + .with_discharge_pressure(400) + .with_energy_function(self.compressor_energy_function.name) + ).validate() + + @property + def fuel_consumption(self): + return LtpTestHelper.consumption_calculate(self, rate=self.fuel_rate, days=[self.days_year2_second_half]) + + @property + def diesel_consumption(self): + return self.consumption_calculate( + rate=self.diesel_rate, days=[self.days_year1_first_half, self.days_year2_first_half] + ) + + @property + def pfs_el_consumption(self): + return self.el_consumption_calculate(power=self.power_usage_mw, days=[self.days_year1_second_half]) + + @property + def gas_turbine_el_generated(self): + return self.el_consumption_calculate( + power=self.power_usage_mw, + days=[self.days_year1_first_half, self.days_year2_first_half, self.days_year2_second_half], + ) + + @property + def boiler_fuel_consumption(self): + return self.consumption_calculate( + rate=self.fuel_rate, + days=[self.days_year1_first_half, self.days_year1_second_half, self.days_year2_first_half], + ) + + @property + def heater_fuel_consumption(self): + return self.consumption_calculate(rate=self.fuel_rate, days=[self.days_year2_second_half]) + + @property + def co2_from_boiler(self): + return self.emission_calculate( + rate=self.fuel_rate, + factor=self.co2_factor, + days=[self.days_year1_first_half, self.days_year1_second_half, self.days_year2_first_half], + ) + + @property + def co2_from_heater(self): + return self.emission_calculate(rate=self.fuel_rate, factor=self.co2_factor, days=[self.days_year2_second_half]) + + @property + def co2_from_fuel(self): + return self.emission_calculate(rate=self.fuel_rate, factor=self.co2_factor, days=[self.days_year2_second_half]) + + @property + def co2_from_diesel(self): + return self.emission_calculate( + rate=self.diesel_rate, + factor=self.co2_factor, + days=[self.days_year1_first_half, self.days_year2_first_half], + ) + + @property + def ch4_from_diesel(self): + return self.emission_calculate( + rate=self.diesel_rate, + factor=self.ch4_factor, + days=[self.days_year1_first_half, self.days_year2_first_half], + ) + + @property + def nox_from_diesel(self): + return self.emission_calculate( + rate=self.diesel_rate, + factor=self.nox_factor, + days=[self.days_year1_first_half, self.days_year2_first_half], + ) + + @property + def nmvoc_from_diesel(self): + return self.emission_calculate( + rate=self.diesel_rate, + factor=self.nmvoc_factor, + days=[self.days_year1_first_half, self.days_year2_first_half], + ) + + @property + def offshore_wind_el_consumption(self): + return self.el_consumption_calculate( + power=self.power_offshore_wind_mw, days=[self.days_year1_second_half, self.days_year2_second_half] + ) * (-1) + + @property + def gas_turbine_compressor_el_consumption(self): + return self.el_consumption_calculate( + power=self.power_compressor_mw, days=[self.days_year1_second_half, self.days_year2_second_half] + ) + + +@pytest.fixture(scope="module") +def ltp_test_helper(): + return LtpTestHelper() + + +class TestLtp: + def test_emissions_diesel_fixed_and_mobile( + self, + request, + ltp_test_helper, + fuel_gas_factory, + diesel_factory, + el_consumer_direct_base_load_factory, + generator_diesel_power_to_fuel_resource, + generator_fuel_power_to_fuel_resource, + ): + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + diesel = diesel_factory( + ["co2", "ch4", "nox", "nmvoc"], + [ + ltp_test_helper.co2_factor, + ltp_test_helper.ch4_factor, + ltp_test_helper.nox_factor, + ltp_test_helper.nmvoc_factor, + ], + ) + + generator_fixed = ( + YamlGeneratorSetBuilder() + .with_name("generator_fixed") + .with_category(ltp_test_helper.category_dict()) + .with_consumers( + [ + el_consumer_direct_base_load_factory( + el_reference_name="base_load", load=ltp_test_helper.load_consumer + ) + ] + ) + .with_electricity2fuel( + ltp_test_helper.temporal_dict( + reference1=ltp_test_helper.generator_diesel_energy_function.name, + reference2=ltp_test_helper.generator_fuel_energy_function.name, + ) + ) + ).validate() + + generator_mobile = deepcopy(generator_fixed) + generator_mobile.name = "generator_mobile" + + installation_fixed = ( + YamlInstallationBuilder() + .with_name("INSTALLATION_FIXED") + .with_regularity(ltp_test_helper.regularity_installation) + .with_fuel(ltp_test_helper.fuel_multi_temporal(diesel, fuel)) + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_generator_sets([generator_fixed]) + ).validate() + + installation_mobile = ( + YamlInstallationBuilder() + .with_name("INSTALLATION_MOBILE") + .with_regularity(ltp_test_helper.regularity_installation) + .with_fuel(ltp_test_helper.fuel_multi_temporal(diesel, fuel)) + .with_category(InstallationUserDefinedCategoryType.MOBILE) + .with_generator_sets([generator_mobile]) + ).validate() + + resources = { + ltp_test_helper.generator_diesel_energy_function.name: generator_diesel_power_to_fuel_resource( + power_usage_mw=ltp_test_helper.power_usage_mw, diesel_rate=ltp_test_helper.diesel_rate ), - installation_compressor_dto( - [simple_direct_el_consumer(name="direct_el_consumer 2")], - installation_name=installation_name_2, - genset_name="generator 2", - compressor_name="gas driven compressor 2", + ltp_test_helper.generator_fuel_energy_function.name: generator_fuel_power_to_fuel_resource( + power_usage_mw=ltp_test_helper.power_usage_mw, fuel_rate=ltp_test_helper.fuel_rate ), - ], - ) - - asset_result = calculate_asset_result(model=asset, variables=variables) - power_electrical_installation_1 = asset_result.get_component_by_name( - installation_name_1 - ).power_electrical_cumulative.values[-1] - - power_mechanical_installation_1 = asset_result.get_component_by_name( - installation_name_1 - ).power_mechanical_cumulative.values[-1] - - power_electrical_installation_2 = asset_result.get_component_by_name( - installation_name_2 - ).power_electrical_cumulative.values[-1] - - power_mechanical_installation_2 = asset_result.get_component_by_name( - installation_name_2 - ).power_mechanical_cumulative.values[-1] - - asset_power_electrical = asset_result.get_component_by_name("Asset 1").power_electrical_cumulative.values[-1] - - asset_power_mechanical = asset_result.get_component_by_name("Asset 1").power_mechanical_cumulative.values[-1] - - # Verify that electrical power is correct at asset level - assert asset_power_electrical == power_electrical_installation_1 + power_electrical_installation_2 - - # Verify that mechanical power is correct at asset level: - assert asset_power_mechanical == power_mechanical_installation_1 + power_mechanical_installation_2 - - -def test_power_from_shore(ltp_pfs_yaml_factory): - """Test power from shore output for LTP export.""" - - time_vector_yearly = pd.date_range(datetime(2025, 1, 1), datetime(2031, 1, 1), freq="YS").to_pydatetime().tolist() - - VariablesMap(time_vector=time_vector_yearly, variables={}) - regularity = 0.2 - load = 10 - cable_loss = 0.1 - max_from_shore = 12 - - dto_case = ltp_pfs_yaml_factory( - regularity=regularity, - cable_loss=cable_loss, - max_usage_from_shore=max_from_shore, - load_direct_consumer=load, - path=Path(ltp_export.__path__[0]), - ) - - dto_case.ecalc_model.model_validate(dto_case.ecalc_model) - - dto_case_csv = ltp_pfs_yaml_factory( - regularity=regularity, - cable_loss="CABLE_LOSS;CABLE_LOSS_FACTOR", - max_usage_from_shore=max_from_shore, - load_direct_consumer=load, - path=Path(ltp_export.__path__[0]), - ) - - ltp_result = get_consumption( - model=dto_case.ecalc_model, variables=dto_case.variables, periods=dto_case.variables.get_periods() - ) - ltp_result_csv = get_consumption( - model=dto_case_csv.ecalc_model, variables=dto_case.variables, periods=dto_case.variables.get_periods() - ) - - power_from_shore_consumption = get_sum_ltp_column( - ltp_result=ltp_result, installation_nr=0, ltp_column="fromShoreConsumption" - ) - power_supply_onshore = get_sum_ltp_column(ltp_result=ltp_result, installation_nr=0, ltp_column="powerSupplyOnshore") - max_usage_from_shore = get_sum_ltp_column( - ltp_result=ltp_result, installation_nr=0, ltp_column="fromShorePeakMaximum" - ) - - power_supply_onshore_csv = get_sum_ltp_column( - ltp_result=ltp_result_csv, installation_nr=0, ltp_column="powerSupplyOnshore" - ) - - # In the temporal model, the category is POWER_FROM_SHORE the last three years, within the period 2025 - 2030: - delta_days = calculate_delta_days(time_vector_yearly)[2:6] - - # Check that power from shore consumption is correct - assert power_from_shore_consumption == sum([load * days * regularity * 24 / 1000 for days in delta_days]) - - # Check that power supply onshore is power from shore consumption * (1 + cable loss) - assert power_supply_onshore == pytest.approx( - sum([load * (1 + cable_loss) * days * regularity * 24 / 1000 for days in delta_days]) - ) - - # Check that max usage from shore is just a report of the input - # Max usage from shore is 0 until 2027.6.1 and the 12 until 2031.1.1, so - # 2027, 2028, 2029 and 2030 (4 years) should all have 12 as max usage from shore. - assert max_usage_from_shore == max_from_shore * 4 - - # Check that reading cable loss from csv-file gives same result as using constant - assert power_supply_onshore == power_supply_onshore_csv - - # Verify correct unit for max usage from shore - assert ltp_result.query_results[0].query_results[3].unit == Unit.MEGA_WATT - - -def test_max_usage_from_shore(ltp_pfs_yaml_factory): - """Test power from shore output for LTP export.""" - - regularity = 0.2 - load = 10 - cable_loss = 0.1 - - dto_case_csv = ltp_pfs_yaml_factory( - regularity=regularity, - cable_loss=cable_loss, - max_usage_from_shore="MAX_USAGE_FROM_SHORE;MAX_USAGE_FROM_SHORE", - load_direct_consumer=load, - path=Path(ltp_export.__path__[0]), - ) - - ltp_result_csv = get_consumption( - model=dto_case_csv.ecalc_model, variables=dto_case_csv.variables, periods=dto_case_csv.variables.get_periods() - ) - - max_usage_from_shore_2027 = float( - ltp_result_csv.query_results[0].query_results[3].values[Period(datetime(2027, 1, 1), datetime(2028, 1, 1))] - ) - - # In the input csv-file max usage from shore is 250 (1.12.2026), 290 (1.6.2027), 283 (1.1.2028) - # and 283 (1.1.2029). Ensure that the correct value is set for 2027 (290 from 1.6): - assert max_usage_from_shore_2027 == 290.0 - - # Ensure that values in 2027, 2028 and 2029 are correct, based on input file: - assert [float(max_pfs) for max_pfs in ltp_result_csv.query_results[0].query_results[3].values.values()][2:5] == [ - 290, - 283, - 283, - ] + } + + asset = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.time_vector_installation[0])) + .with_end(str(ltp_test_helper.time_vector_installation[-1])) + .with_installations([installation_fixed, installation_mobile]) + .with_facility_inputs( + [ltp_test_helper.generator_diesel_energy_function, ltp_test_helper.generator_fuel_energy_function] + ) + .with_fuel_types([fuel, diesel]) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources=resources, frequency=Frequency.YEAR) + + variables = ltp_test_helper.create_variables_map(ltp_test_helper.time_vector_installation, [1, 1, 1, 1]) + ltp_result = ltp_test_helper.get_ltp_result(asset, variables) + + ltp_test_helper.assert_emissions(ltp_result, 0, "engineDieselCo2Mass", ltp_test_helper.co2_from_diesel) + ltp_test_helper.assert_emissions(ltp_result, 1, "engineNoCo2TaxDieselCo2Mass", ltp_test_helper.co2_from_diesel) + ltp_test_helper.assert_emissions(ltp_result, 0, "engineDieselNoxMass", ltp_test_helper.nox_from_diesel) + ltp_test_helper.assert_emissions(ltp_result, 1, "engineNoCo2TaxDieselNoxMass", ltp_test_helper.nox_from_diesel) + ltp_test_helper.assert_emissions(ltp_result, 0, "engineDieselNmvocMass", ltp_test_helper.nmvoc_from_diesel) + ltp_test_helper.assert_emissions( + ltp_result, 1, "engineNoCo2TaxDieselNmvocMass", ltp_test_helper.nmvoc_from_diesel + ) + ltp_test_helper.assert_emissions(ltp_result, 0, "engineDieselCh4Mass", ltp_test_helper.ch4_from_diesel) + ltp_test_helper.assert_emissions(ltp_result, 1, "engineNoCo2TaxDieselCh4Mass", ltp_test_helper.ch4_from_diesel) + + def test_temporal_models_detailed( + self, + request, + ltp_test_helper, + diesel_factory, + fuel_gas_factory, + generator_diesel_power_to_fuel_resource, + generator_fuel_power_to_fuel_resource, + el_consumer_direct_base_load_factory, + ): + """Test various queries for LTP reporting. Purpose: ensure that variations in temporal models are captured. + + Detailed temporal models (variations within one year) for: + - Fuel type + - Generator set user defined category + - Generator set model + """ + variables = ltp_test_helper.create_variables_map( + ltp_test_helper.time_vector_installation, rate_values=[1, 1, 1, 1] + ) + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + diesel = diesel_factory(["co2"], [ltp_test_helper.co2_factor]) + + generator_set = ( + YamlGeneratorSetBuilder() + .with_name("generator_set") + .with_category(ltp_test_helper.category_dict_coarse()) + .with_consumers( + [ + el_consumer_direct_base_load_factory( + el_reference_name="base_load", load=ltp_test_helper.load_consumer + ) + ] + ) + .with_electricity2fuel( + ltp_test_helper.temporal_dict( + reference1=ltp_test_helper.generator_diesel_energy_function.name, + reference2=ltp_test_helper.generator_fuel_energy_function.name, + ) + ) + ).validate() + + installation = ( + YamlInstallationBuilder() + .with_name("INSTALLATION A") + .with_generator_sets([generator_set]) + .with_fuel(ltp_test_helper.fuel_multi_temporal(fuel1=diesel, fuel2=fuel)) + .with_category(InstallationUserDefinedCategoryType.FIXED) + ).validate() + + resources = { + ltp_test_helper.generator_diesel_energy_function.name: generator_diesel_power_to_fuel_resource( + power_usage_mw=ltp_test_helper.power_usage_mw, + diesel_rate=ltp_test_helper.diesel_rate, + ), + ltp_test_helper.generator_fuel_energy_function.name: generator_fuel_power_to_fuel_resource( + power_usage_mw=ltp_test_helper.power_usage_mw, + fuel_rate=ltp_test_helper.fuel_rate, + ), + } + + asset = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.time_vector_installation[0])) + .with_end(str(ltp_test_helper.time_vector_installation[-1])) + .with_installations([installation]) + .with_facility_inputs( + [ltp_test_helper.generator_diesel_energy_function, ltp_test_helper.generator_fuel_energy_function] + ) + .with_fuel_types([fuel, diesel]) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources=resources, frequency=Frequency.YEAR) + + ltp_result = ltp_test_helper.get_ltp_result(asset, variables) + + turbine_fuel_consumption = ltp_test_helper.get_sum_ltp_column( + ltp_result, installation_nr=0, ltp_column="turbineFuelGasConsumption" + ) + + # FuelQuery: Check that turbine fuel consumption is included, + # even if the temporal model starts with diesel every year + assert turbine_fuel_consumption != 0 + + ltp_test_helper.assert_consumption(ltp_result, 0, "turbineFuelGasConsumption", ltp_test_helper.fuel_consumption) + ltp_test_helper.assert_consumption(ltp_result, 0, "engineDieselConsumption", ltp_test_helper.diesel_consumption) + ltp_test_helper.assert_consumption(ltp_result, 0, "turbineFuelGasCo2Mass", ltp_test_helper.co2_from_fuel) + ltp_test_helper.assert_consumption(ltp_result, 0, "engineDieselCo2Mass", ltp_test_helper.co2_from_diesel) + ltp_test_helper.assert_consumption(ltp_result, 0, "fromShoreConsumption", ltp_test_helper.pfs_el_consumption) + ltp_test_helper.assert_consumption( + ltp_result, 0, "gasTurbineGeneratorConsumption", ltp_test_helper.gas_turbine_el_generated + ) + + def test_temporal_models_offshore_wind( + self, + request, + ltp_test_helper, + fuel_gas_factory, + generator_fuel_power_to_fuel_resource, + ): + """Test ElConsumerPowerConsumptionQuery for calculating offshore wind el-consumption, LTP. + + Detailed temporal models (variations within one year) for: + - El-consumer user defined category + - El-consumer energy usage model + """ + variables = ltp_test_helper.create_variables_map( + ltp_test_helper.time_vector_installation, rate_values=[1, 1, 1, 1] + ) + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + + generator_set = ( + YamlGeneratorSetBuilder() + .with_name("generator_set") + .with_category({ltp_test_helper.period_from_date1.start: ConsumerUserDefinedCategoryType.TURBINE_GENERATOR}) + .with_consumers([ltp_test_helper.offshore_wind_consumer(request, ltp_test_helper.power_offshore_wind_mw)]) + .with_electricity2fuel( + {ltp_test_helper.period_from_date1.start: ltp_test_helper.generator_fuel_energy_function.name} + ) + ).validate() + + installation = ( + YamlInstallationBuilder() + .with_name("INSTALLATION A") + .with_generator_sets([generator_set]) + .with_fuel(fuel.name) + .with_category(InstallationUserDefinedCategoryType.FIXED) + ).validate() + + resources = { + ltp_test_helper.generator_fuel_energy_function.name: generator_fuel_power_to_fuel_resource( + power_usage_mw=ltp_test_helper.power_usage_mw, + fuel_rate=ltp_test_helper.fuel_rate, + ), + } + + asset = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.time_vector_installation[0])) + .with_end(str(ltp_test_helper.time_vector_installation[-1])) + .with_installations([installation]) + .with_facility_inputs([ltp_test_helper.generator_fuel_energy_function]) + .with_fuel_types([fuel]) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources=resources, frequency=Frequency.YEAR) + + ltp_result = ltp_test_helper.get_ltp_result(asset, variables) + + offshore_wind_el_consumption = ltp_test_helper.get_sum_ltp_column( + ltp_result, installation_nr=0, ltp_column="offshoreWindConsumption" + ) + + # ElConsumerPowerConsumptionQuery: Check that offshore wind el-consumption is correct. + assert offshore_wind_el_consumption == ltp_test_helper.offshore_wind_el_consumption + + def test_temporal_models_compressor( + self, + request, + ltp_test_helper, + generator_fuel_power_to_fuel_resource, + compressor_sampled_fuel_driven_resource, + fuel_gas_factory, + ): + """Test FuelConsumerPowerConsumptionQuery for calculating gas turbine compressor el-consumption, LTP. + + Detailed temporal models (variations within one year) for: + - Fuel consumer user defined category + """ + variables = ltp_test_helper.create_variables_map( + ltp_test_helper.time_vector_installation, rate_values=[1, 1, 1, 1] + ) + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + + installation = ( + YamlInstallationBuilder() + .with_name("INSTALLATION A") + .with_fuel(fuel.name) + .with_fuel_consumers([ltp_test_helper.fuel_consumer_compressor(fuel.name)]) + .with_category(InstallationUserDefinedCategoryType.FIXED) + ).validate() + + resources = { + ltp_test_helper.compressor_energy_function.name: compressor_sampled_fuel_driven_resource( + compressor_rate=ltp_test_helper.compressor_rate, power_compressor_mw=ltp_test_helper.power_compressor_mw + ) + } + + asset = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.time_vector_installation[0])) + .with_end(str(ltp_test_helper.time_vector_installation[-1])) + .with_installations([installation]) + .with_facility_inputs([ltp_test_helper.compressor_energy_function]) + .with_fuel_types([fuel]) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources=resources, frequency=Frequency.YEAR) + + ltp_result = ltp_test_helper.get_ltp_result(asset, variables) + + gas_turbine_compressor_el_consumption = ltp_test_helper.get_sum_ltp_column( + ltp_result, installation_nr=0, ltp_column="gasTurbineCompressorConsumption" + ) + + # FuelConsumerPowerConsumptionQuery. Check gas turbine compressor el consumption. + assert gas_turbine_compressor_el_consumption == ltp_test_helper.gas_turbine_compressor_el_consumption + + def test_boiler_heater_categories(self, request, ltp_test_helper, fuel_gas_factory): + variables = ltp_test_helper.create_variables_map(ltp_test_helper.time_vector_installation) + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + + energy_usage_model = ( + YamlEnergyUsageModelDirectBuilder() + .with_fuel_rate(ltp_test_helper.fuel_rate) + .with_consumption_rate_type(ConsumptionRateType.STREAM_DAY) + ).validate() + + fuel_consumer = ( + YamlFuelConsumerBuilder() + .with_name("boiler_heater") + .with_fuel(fuel.name) + .with_energy_usage_model({ltp_test_helper.full_period.start: energy_usage_model}) + .with_category( + { + Period(ltp_test_helper.date1, ltp_test_helper.date4).start: ConsumerUserDefinedCategoryType.BOILER, + Period(ltp_test_helper.date4).start: ConsumerUserDefinedCategoryType.HEATER, + } + ) + ).validate() + + installation = ( + YamlInstallationBuilder() + .with_name("INSTALLATION A") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_fuel(fuel.name) + .with_fuel_consumers([fuel_consumer]) + .with_regularity(ltp_test_helper.regularity_installation) + ).validate() + + asset = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.date1)) + .with_end(str(ltp_test_helper.date5)) + .with_installations([installation]) + .with_fuel_types([fuel]) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources={}, frequency=Frequency.YEAR) + + ltp_result = ltp_test_helper.get_ltp_result(asset, variables) + + ltp_test_helper.assert_consumption( + ltp_result, 0, "boilerFuelGasConsumption", ltp_test_helper.boiler_fuel_consumption + ) + ltp_test_helper.assert_consumption( + ltp_result, 0, "heaterFuelGasConsumption", ltp_test_helper.heater_fuel_consumption + ) + ltp_test_helper.assert_consumption(ltp_result, 0, "boilerFuelGasCo2Mass", ltp_test_helper.co2_from_boiler) + ltp_test_helper.assert_consumption(ltp_result, 0, "heaterFuelGasCo2Mass", ltp_test_helper.co2_from_heater) + + def test_total_oil_loaded_old_method( + self, request, ltp_test_helper, fuel_gas_factory, fuel_consumer_direct_factory + ): + """Test total oil loaded/stored for LTP export. Using original method where direct/venting emitters are + modelled as FUELSCONSUMERS using DIRECT. + + Verify correct volume when model includes emissions related to both storage and loading of oil, + and when model includes only loading. + """ + time_vector = [datetime(2027, 1, 1), datetime(2028, 1, 1)] + variables = ltp_test_helper.create_variables_map(time_vector) + + regularity = 0.6 + emission_factor = 2 + rate = 100 + + fuel = fuel_gas_factory(["ch4"], [emission_factor]) + loading = fuel_consumer_direct_factory( + fuel_reference_name=fuel.name, rate=rate, name="loading", category=ConsumerUserDefinedCategoryType.LOADING + ) + + storage = fuel_consumer_direct_factory( + fuel_reference_name=fuel.name, rate=rate, name="storage", category=ConsumerUserDefinedCategoryType.STORAGE + ) + + installation = ( + YamlInstallationBuilder() + .with_name("minimal_installation") + .with_fuel(fuel.name) + .with_fuel_consumers([loading, storage]) + .with_regularity(regularity) + .with_category(InstallationUserDefinedCategoryType.FIXED) + ).validate() + + asset = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.time_vector_installation[0])) + .with_end(str(ltp_test_helper.time_vector_installation[-1])) + .with_installations([installation]) + .with_fuel_types([fuel]) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources={}, frequency=Frequency.YEAR) + + installation_loading_only = ( + YamlInstallationBuilder() + .with_name("minimal_installation") + .with_fuel(fuel.name) + .with_fuel_consumers([loading]) + .with_regularity(regularity) + .with_category(InstallationUserDefinedCategoryType.FIXED) + ).validate() + + asset_loading_only = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.time_vector_installation[0])) + .with_end(str(ltp_test_helper.time_vector_installation[-1])) + .with_installations([installation_loading_only]) + .with_fuel_types([fuel]) + ).validate() + + asset_loading_only = ltp_test_helper.get_yaml_model( + request, asset=asset_loading_only, resources={}, frequency=Frequency.YEAR + ) + + ltp_result_loading_storage = ltp_test_helper.get_ltp_result(asset, variables) + ltp_result_loading_only = ltp_test_helper.get_ltp_result(asset_loading_only, variables) + + loaded_and_stored_oil_loading_and_storage = ltp_test_helper.get_sum_ltp_column( + ltp_result_loading_storage, installation_nr=0, ltp_column="loadedAndStoredOil" + ) + loaded_and_stored_oil_loading_only = ltp_test_helper.get_sum_ltp_column( + ltp_result_loading_only, installation_nr=0, ltp_column="loadedAndStoredOil" + ) + + # Verify output for total oil loaded/stored, if only loading is specified. + assert loaded_and_stored_oil_loading_only is not None + + # Verify correct volume for oil loaded/stored + assert loaded_and_stored_oil_loading_and_storage == rate * 365 * regularity + + # Verify that total oil loaded/stored is the same if only loading is specified, + # compared to a model with both loading and storage. + assert loaded_and_stored_oil_loading_and_storage == loaded_and_stored_oil_loading_only + + def test_electrical_and_mechanical_power_installation( + self, + ltp_test_helper, + request, + fuel_gas_factory, + generator_fuel_power_to_fuel_resource, + compressor_sampled_fuel_driven_resource, + ): + """Check that new total power includes the sum of electrical- and mechanical power at installation level""" + variables = ltp_test_helper.create_variables_map(ltp_test_helper.time_vector_installation) + + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + + generator_set = ltp_test_helper.generator_set(request) + + installation = ( + YamlInstallationBuilder() + .with_name("INSTALLATION A") + .with_generator_sets([generator_set]) + .with_fuel(fuel.name) + .with_regularity(ltp_test_helper.regularity_installation) + .with_fuel_consumers([ltp_test_helper.fuel_consumer_compressor(fuel.name)]) + .with_category(InstallationUserDefinedCategoryType.FIXED) + ).validate() + + resources = { + ltp_test_helper.generator_fuel_energy_function.name: generator_fuel_power_to_fuel_resource( + power_usage_mw=ltp_test_helper.power_usage_mw, + fuel_rate=ltp_test_helper.fuel_rate, + ), + ltp_test_helper.compressor_energy_function.name: compressor_sampled_fuel_driven_resource( + compressor_rate=ltp_test_helper.compressor_rate, power_compressor_mw=ltp_test_helper.power_compressor_mw + ), + } + + asset = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.time_vector_installation[0])) + .with_end(str(ltp_test_helper.time_vector_installation[-1])) + .with_installations([installation]) + .with_fuel_types([fuel]) + .with_facility_inputs( + [ltp_test_helper.generator_fuel_energy_function, ltp_test_helper.compressor_energy_function] + ) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources=resources, frequency=Frequency.YEAR) + + asset_result = ltp_test_helper.calculate_asset_result(model=asset, variables=variables) + power_fuel_driven_compressor = asset_result.get_component_by_name("compressor").power_cumulative.values[-1] + power_generator_set = asset_result.get_component_by_name("generator_set").power_cumulative.values[-1] + + # Extract cumulative electrical-, mechanical- and total power. + power_electrical_installation = asset_result.get_component_by_name( + "INSTALLATION A" + ).power_electrical_cumulative.values[-1] + + power_mechanical_installation = asset_result.get_component_by_name( + "INSTALLATION A" + ).power_mechanical_cumulative.values[-1] + + power_total_installation = asset_result.get_component_by_name("INSTALLATION A").power_cumulative.values[-1] + + # Verify that total power is correct + assert power_total_installation == power_electrical_installation + power_mechanical_installation + + # Verify that electrical power equals genset power, and mechanical power equals power from gas driven compressor: + assert power_generator_set == power_electrical_installation + assert power_fuel_driven_compressor == power_mechanical_installation + + def test_electrical_and_mechanical_power_asset( + self, + ltp_test_helper, + request, + fuel_gas_factory, + el_consumer_direct_base_load_factory, + generator_fuel_power_to_fuel_resource, + compressor_sampled_fuel_driven_resource, + ): + """Check that new total power includes the sum of electrical- and mechanical power at installation level""" + variables = ltp_test_helper.create_variables_map(ltp_test_helper.time_vector_installation) + name1 = "INSTALLATION_1" + name2 = "INSTALLATION_2" + + el_consumer1 = el_consumer_direct_base_load_factory( + el_reference_name="base_load1", load=ltp_test_helper.load_consumer + ) + el_consumer2 = el_consumer_direct_base_load_factory( + el_reference_name="base_load2", load=ltp_test_helper.load_consumer + ) + + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + + generator_set1 = ltp_test_helper.generator_set(request, name="generator_set1", el_consumer=el_consumer1) + generator_set2 = ltp_test_helper.generator_set(request, name="generator_set2", el_consumer=el_consumer2) + + installation1 = ( + YamlInstallationBuilder() + .with_name(name1) + .with_generator_sets([generator_set1]) + .with_fuel(fuel.name) + .with_regularity(ltp_test_helper.regularity_installation) + .with_fuel_consumers([ltp_test_helper.fuel_consumer_compressor(fuel.name)]) + .with_category(InstallationUserDefinedCategoryType.FIXED) + ).validate() + + installation2 = ( + YamlInstallationBuilder() + .with_name(name2) + .with_generator_sets([generator_set2]) + .with_fuel(fuel.name) + .with_regularity(ltp_test_helper.regularity_installation) + .with_fuel_consumers([ltp_test_helper.fuel_consumer_compressor(fuel.name, name="compressor2")]) + .with_category(InstallationUserDefinedCategoryType.FIXED) + ).validate() + + resources = { + ltp_test_helper.generator_fuel_energy_function.name: generator_fuel_power_to_fuel_resource( + power_usage_mw=ltp_test_helper.power_usage_mw, + fuel_rate=ltp_test_helper.fuel_rate, + ), + ltp_test_helper.compressor_energy_function.name: compressor_sampled_fuel_driven_resource( + compressor_rate=ltp_test_helper.compressor_rate, power_compressor_mw=ltp_test_helper.power_compressor_mw + ), + } + + asset = ( + YamlAssetBuilder() + .with_start(str(ltp_test_helper.time_vector_installation[0])) + .with_end(str(ltp_test_helper.time_vector_installation[-1])) + .with_installations([installation1, installation2]) + .with_fuel_types([fuel]) + .with_facility_inputs( + [ltp_test_helper.generator_fuel_energy_function, ltp_test_helper.compressor_energy_function] + ) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources=resources, frequency=Frequency.YEAR) + + asset_result = ltp_test_helper.calculate_asset_result(model=asset, variables=variables) + + power_el_1 = asset_result.get_component_by_name(name1).power_electrical_cumulative.values[-1] + power_mech_1 = asset_result.get_component_by_name(name1).power_mechanical_cumulative.values[-1] + + power_el_2 = asset_result.get_component_by_name(name2).power_electrical_cumulative.values[-1] + power_mech_2 = asset_result.get_component_by_name(name2).power_mechanical_cumulative.values[-1] + + asset_power_el = asset_result.component_result.power_electrical_cumulative.values[-1] + + asset_power_mech = asset_result.component_result.power_mechanical_cumulative.values[-1] + + # Verify that electrical power is correct at asset level + assert asset_power_el == power_el_1 + power_el_2 + + # Verify that mechanical power is correct at asset level: + assert asset_power_mech == power_mech_1 + power_mech_2 + + def test_venting_emitters(self, request, ltp_test_helper, fuel_consumer_direct_factory, fuel_gas_factory): + """Test venting emitters for LTP export. + + Verify correct behaviour if input rate is given in different units and rate types (sd and cd). + """ + + time_vector = [datetime(2027, 1, 1), datetime(2028, 1, 1)] + regularity = 0.2 + emission_rate = 10 + + variables = ltp_test_helper.create_variables_map(time_vector) + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + + venting_emitter_sd_kg_per_day = ( + YamlVentingEmitterDirectTypeBuilder() + .with_name("Venting emitter 1") + .with_category(ConsumerUserDefinedCategoryType.STORAGE) + .with_emission_names_rates_units_and_types( + names=["ch4"], + rates=[emission_rate], + units=[YamlEmissionRateUnits.KILO_PER_DAY], + rate_types=[RateType.STREAM_DAY], + ) + ).validate() + + venting_emitter_sd_tons_per_day = ( + YamlVentingEmitterDirectTypeBuilder() + .with_name("Venting emitter 1") + .with_category(ConsumerUserDefinedCategoryType.STORAGE) + .with_emission_names_rates_units_and_types( + names=["ch4"], + rates=[emission_rate], + units=[YamlEmissionRateUnits.TONS_PER_DAY], + rate_types=[RateType.STREAM_DAY], + ) + ).validate() + + venting_emitter_cd_kg_per_day = ( + YamlVentingEmitterDirectTypeBuilder() + .with_name("Venting emitter 1") + .with_category(ConsumerUserDefinedCategoryType.STORAGE) + .with_emission_names_rates_units_and_types( + names=["ch4"], + rates=[emission_rate], + units=[YamlEmissionRateUnits.KILO_PER_DAY], + rate_types=[RateType.CALENDAR_DAY], + ) + ).validate() + + installation_sd_kg_per_day = ( + YamlInstallationBuilder() + .with_name("minimal_installation") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_fuel_consumers([fuel_consumer_direct_factory(fuel.name, ltp_test_helper.fuel_rate)]) + .with_venting_emitters([venting_emitter_sd_kg_per_day]) + .with_regularity(regularity) + ).validate() + + installation_sd_tons_per_day = ( + YamlInstallationBuilder() + .with_name("minimal_installation") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_fuel_consumers([fuel_consumer_direct_factory(fuel.name, ltp_test_helper.fuel_rate)]) + .with_venting_emitters([venting_emitter_sd_tons_per_day]) + .with_regularity(regularity) + ).validate() + + installation_cd_kg_per_day = ( + YamlInstallationBuilder() + .with_name("minimal_installation") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_fuel_consumers([fuel_consumer_direct_factory(fuel.name, ltp_test_helper.fuel_rate)]) + .with_venting_emitters([venting_emitter_cd_kg_per_day]) + .with_regularity(regularity) + ).validate() + + asset_sd_kg_per_day = ( + YamlAssetBuilder() + .with_installations([installation_sd_kg_per_day]) + .with_fuel_types([fuel]) + .with_start(str(time_vector[0])) + .with_end(str(time_vector[-1])) + ).validate() + + asset_sd_tons_per_day = ( + YamlAssetBuilder() + .with_installations([installation_sd_tons_per_day]) + .with_fuel_types([fuel]) + .with_start(str(time_vector[0])) + .with_end(str(time_vector[-1])) + ).validate() + + asset_cd_kg_per_day = ( + YamlAssetBuilder() + .with_installations([installation_cd_kg_per_day]) + .with_fuel_types([fuel]) + .with_start(str(time_vector[0])) + .with_end(str(time_vector[-1])) + ).validate() + + asset_sd_kg_per_day = ltp_test_helper.get_yaml_model( + request, asset=asset_sd_kg_per_day, resources={}, frequency=Frequency.YEAR + ) + asset_sd_tons_per_day = ltp_test_helper.get_yaml_model( + request, asset=asset_sd_tons_per_day, resources={}, frequency=Frequency.YEAR + ) + asset_cd_kg_per_day = ltp_test_helper.get_yaml_model( + request, asset=asset_cd_kg_per_day, resources={}, frequency=Frequency.YEAR + ) + + ltp_result_input_sd_kg_per_day = ltp_test_helper.get_ltp_result(asset_sd_kg_per_day, variables) + ltp_result_input_sd_tons_per_day = ltp_test_helper.get_ltp_result(asset_sd_tons_per_day, variables) + ltp_result_input_cd_kg_per_day = ltp_test_helper.get_ltp_result(asset_cd_kg_per_day, variables) + + emission_input_sd_kg_per_day = ltp_test_helper.get_sum_ltp_column( + ltp_result_input_sd_kg_per_day, installation_nr=0, ltp_column="storageCh4Mass" + ) + emission_input_sd_tons_per_day = ltp_test_helper.get_sum_ltp_column( + ltp_result_input_sd_tons_per_day, installation_nr=0, ltp_column="storageCh4Mass" + ) + emission_input_cd_kg_per_day = ltp_test_helper.get_sum_ltp_column( + ltp_result_input_cd_kg_per_day, installation_nr=0, ltp_column="storageCh4Mass" + ) + + # Verify correct emissions when input is kg per day. Output should be in tons per day - hence dividing by 1000 + assert emission_input_sd_kg_per_day == (emission_rate / 1000) * 365 * regularity + + # Verify correct emissions when input is tons per day. + assert emission_input_sd_tons_per_day == emission_rate * 365 * regularity + + # Verify that input calendar day vs input stream day is linked correctly through regularity + assert emission_input_cd_kg_per_day == emission_input_sd_kg_per_day / regularity + + # Verify that results is independent of regularity, when input rate is in calendar days + assert emission_input_cd_kg_per_day == (emission_rate / 1000) * 365 + + def test_only_venting_emitters_no_fuelconsumers( + self, request, ltp_test_helper, fuel_consumer_direct_factory, fuel_gas_factory + ): + """ + Test that it is possible with only venting emitters, without fuelconsumers. + """ + time_vector = [datetime(2027, 1, 1), datetime(2028, 1, 1)] + regularity = 0.2 + emission_rate = 10 + + variables = ltp_test_helper.create_variables_map(time_vector) + fuel = fuel_gas_factory(["co2"], [ltp_test_helper.co2_factor]) + + # Installation with only venting emitters: + venting_emitter = ( + YamlVentingEmitterDirectTypeBuilder() + .with_name("Venting emitter 1") + .with_category(ConsumerUserDefinedCategoryType.STORAGE) + .with_emission_names_rates_units_and_types( + names=["ch4"], + rates=[emission_rate], + units=[YamlEmissionRateUnits.KILO_PER_DAY], + rate_types=[RateType.STREAM_DAY], + ) + ).validate() + + installation_only_emitters = ( + YamlInstallationBuilder() + .with_name("Venting emitter installation") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_venting_emitters([venting_emitter]) + .with_regularity(regularity) + ).validate() + + asset = ( + YamlAssetBuilder() + .with_installations([installation_only_emitters]) + .with_start(str(time_vector[0])) + .with_end(str(time_vector[-1])) + ).validate() + + asset = ltp_test_helper.get_yaml_model(request, asset=asset, resources={}, frequency=Frequency.YEAR) + + venting_emitter_only_results = ltp_test_helper.get_ltp_result(asset, variables) + + # Verify that eCalc is not failing in get_asset_result with only venting emitters - + # when installation result is empty, i.e. with no genset and fuel consumers: + assert isinstance(ltp_test_helper.calculate_asset_result(model=asset, variables=variables), EcalcModelResult) + + # Verify correct emissions: + emissions_ch4 = ltp_test_helper.get_sum_ltp_column( + venting_emitter_only_results, installation_nr=0, ltp_column="storageCh4Mass" + ) + assert emissions_ch4 == (emission_rate / 1000) * 365 * regularity + + # Installation with only fuel consumers: + installation_only_fuel_consumers = ( + YamlInstallationBuilder() + .with_name("Fuel consumer installation") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_fuel_consumers( + [fuel_consumer_direct_factory(fuel_reference_name=fuel.name, rate=ltp_test_helper.fuel_rate)] + ) + .with_regularity(regularity) + ).validate() + + asset_multi_installations = ( + YamlAssetBuilder() + .with_installations([installation_only_emitters, installation_only_fuel_consumers]) + .with_fuel_types([fuel]) + .with_start(str(time_vector[0])) + .with_end(str(time_vector[-1])) + ).validate() + + asset_multi_installations = ltp_test_helper.get_yaml_model( + request, asset=asset_multi_installations, resources={}, frequency=Frequency.YEAR + ) + + # Verify that eCalc is not failing in get_asset_result, with only venting emitters - + # when installation result is empty for one installation, i.e. with no genset and fuel consumers. + # Include asset with two installations, one with only emitters and one with only fuel consumers - + # ensure that get_asset_result returns a result: + assert isinstance( + ltp_test_helper.calculate_asset_result(model=asset_multi_installations, variables=variables), + EcalcModelResult, + ) + + asset_ltp_result = ltp_test_helper.get_ltp_result(asset_multi_installations, variables) + + # Check that the results are the same: For the case with only one installation (only venting emitters), + # compared to the multi-installation case with two installations. The fuel-consumer installation should + # give no CH4-contribution (only CO2) + emissions_ch4_asset = ltp_test_helper.get_sum_ltp_column( + asset_ltp_result, installation_nr=0, ltp_column="storageCh4Mass" + ) + assert emissions_ch4 == emissions_ch4_asset + + def test_power_from_shore( + self, + request, + ltp_test_helper, + yaml_model_factory, + el_consumer_direct_base_load_factory, + fuel_gas_factory, + generator_electricity2fuel_17MW_resource, + onshore_power_electricity2fuel_resource, + cable_loss_time_series_resource, + ): + """Test power from shore output for LTP export.""" + + time_vector_yearly = ( + pd.date_range(datetime(2025, 1, 1), datetime(2031, 1, 1), freq="YS").to_pydatetime().tolist() + ) + fuel = fuel_gas_factory( + ["co2", "ch4", "nmvoc", "nox"], + [ + ltp_test_helper.co2_factor, + ltp_test_helper.ch4_factor, + ltp_test_helper.nmvoc_factor, + ltp_test_helper.nox_factor, + ], + ) + + regularity = 0.2 + load = 10 + cable_loss = 0.1 + max_from_shore = 12 + + cable_loss_time_series = ( + YamlTimeSeriesBuilder().with_name("CABLE_LOSS").with_type("DEFAULT").with_file("CABLE_LOSS") + ).validate() + + generator_set = ( + YamlGeneratorSetBuilder() + .with_name("generator1") + .with_category( + { + datetime(2025, 1, 1): ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, + datetime(2027, 1, 1): ConsumerUserDefinedCategoryType.POWER_FROM_SHORE, + } + ) + .with_electricity2fuel( + { + datetime(2025, 1, 1): ltp_test_helper.generator_fuel_energy_function.name, + datetime(2027, 1, 1): ltp_test_helper.power_from_shore_energy_function.name, + } + ) + .with_consumers([el_consumer_direct_base_load_factory(el_reference_name="base_load", load=load)]) + .with_cable_loss(cable_loss) + .with_max_usage_from_shore(max_from_shore) + ).validate() + + installation_pfs = ( + YamlInstallationBuilder() + .with_name("minimal_installation") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_fuel(fuel.name) + .with_generator_sets([generator_set]) + .with_regularity(regularity) + ).validate() + + resources = { + ltp_test_helper.generator_fuel_energy_function.name: generator_electricity2fuel_17MW_resource, + ltp_test_helper.power_from_shore_energy_function.name: onshore_power_electricity2fuel_resource, + cable_loss_time_series.name: cable_loss_time_series_resource, + } + + asset_pfs = ( + YamlAssetBuilder() + .with_start(time_vector_yearly[0]) + .with_end(time_vector_yearly[-1]) + .with_installations([installation_pfs]) + .with_fuel_types([fuel]) + .with_time_series([cable_loss_time_series]) + .with_facility_inputs( + [ltp_test_helper.generator_fuel_energy_function, ltp_test_helper.power_from_shore_energy_function] + ) + ).validate() + + asset_pfs = ltp_test_helper.get_yaml_model( + request, asset=asset_pfs, resources=resources, frequency=Frequency.YEAR + ) + generator_set_csv = deepcopy(generator_set) + generator_set_csv.cable_loss = "CABLE_LOSS;CABLE_LOSS_FACTOR" + + installation_pfs_csv = ( + YamlInstallationBuilder() + .with_name("minimal_installation_csv") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_fuel(fuel.name) + .with_generator_sets([generator_set_csv]) + .with_regularity(regularity) + ).validate() + + asset_pfs_csv = ( + YamlAssetBuilder() + .with_start(time_vector_yearly[0]) + .with_end(time_vector_yearly[-1]) + .with_installations([installation_pfs_csv]) + .with_fuel_types([fuel]) + .with_time_series([cable_loss_time_series]) + .with_facility_inputs( + [ltp_test_helper.generator_fuel_energy_function, ltp_test_helper.power_from_shore_energy_function] + ) + ).validate() + + asset_pfs_csv = ltp_test_helper.get_yaml_model( + request, asset=asset_pfs_csv, resources=resources, frequency=Frequency.YEAR + ) + ltp_result = ltp_test_helper.get_ltp_result(asset_pfs, asset_pfs.variables) + ltp_result_csv = ltp_test_helper.get_ltp_result(asset_pfs_csv, asset_pfs_csv.variables) + + power_from_shore_consumption = ltp_test_helper.get_sum_ltp_column( + ltp_result=ltp_result, installation_nr=0, ltp_column="fromShoreConsumption" + ) + power_supply_onshore = ltp_test_helper.get_sum_ltp_column( + ltp_result=ltp_result, installation_nr=0, ltp_column="powerSupplyOnshore" + ) + max_usage_from_shore = ltp_test_helper.get_sum_ltp_column( + ltp_result=ltp_result, installation_nr=0, ltp_column="fromShorePeakMaximum" + ) + + power_supply_onshore_csv = ltp_test_helper.get_sum_ltp_column( + ltp_result=ltp_result_csv, installation_nr=0, ltp_column="powerSupplyOnshore" + ) + + # In the temporal model, the category is POWER_FROM_SHORE the last three years, within the period 2025 - 2030: + delta_days = calculate_delta_days(time_vector_yearly)[2:6] + + # Check that power from shore consumption is correct + assert power_from_shore_consumption == sum([load * days * regularity * 24 / 1000 for days in delta_days]) + + # Check that power supply onshore is power from shore consumption * (1 + cable loss) + assert power_supply_onshore == pytest.approx( + sum([load * (1 + cable_loss) * days * regularity * 24 / 1000 for days in delta_days]) + ) + + # Check that max usage from shore is just a report of the input + # Max usage from shore is 0 until 2027.6.1 and the 12 until 2031.1.1, so + # 2027, 2028, 2029 and 2030 (4 years) should all have 12 as max usage from shore. + assert max_usage_from_shore == max_from_shore * 4 + + # Check that reading cable loss from csv-file gives same result as using constant + assert power_supply_onshore == power_supply_onshore_csv + + # Verify correct unit for max usage from shore + assert ( + ltp_test_helper.get_ltp_column( + ltp_result=ltp_result, installation_nr=0, ltp_column="fromShorePeakMaximum" + ).unit + == Unit.MEGA_WATT + ) + + def test_max_usage_from_shore( + self, + request, + ltp_test_helper, + el_consumer_direct_base_load_factory, + generator_electricity2fuel_17MW_resource, + onshore_power_electricity2fuel_resource, + max_usage_from_shore_time_series_resource, + fuel_gas_factory, + ): + """Test power from shore output for LTP export.""" + + regularity = 0.2 + load = 10 + cable_loss = 0.1 + + time_vector_yearly = ( + pd.date_range(datetime(2025, 1, 1), datetime(2031, 1, 1), freq="YS").to_pydatetime().tolist() + ) + + fuel = fuel_gas_factory( + ["co2", "ch4", "nmvoc", "nox"], + [ + ltp_test_helper.co2_factor, + ltp_test_helper.ch4_factor, + ltp_test_helper.nmvoc_factor, + ltp_test_helper.nox_factor, + ], + ) + + max_usage_from_shore_time_series = ( + YamlTimeSeriesBuilder() + .with_name("MAX_USAGE_FROM_SHORE") + .with_type("DEFAULT") + .with_file("MAX_USAGE_FROM_SHORE") + ).validate() + + generator_set = ( + YamlGeneratorSetBuilder() + .with_name("generator1") + .with_category( + { + datetime(2025, 1, 1): ConsumerUserDefinedCategoryType.TURBINE_GENERATOR, + datetime(2027, 1, 1): ConsumerUserDefinedCategoryType.POWER_FROM_SHORE, + } + ) + .with_electricity2fuel( + { + datetime(2025, 1, 1): ltp_test_helper.generator_fuel_energy_function.name, + datetime(2027, 1, 1): ltp_test_helper.power_from_shore_energy_function.name, + } + ) + .with_consumers([el_consumer_direct_base_load_factory(el_reference_name="base_load", load=load)]) + .with_cable_loss(cable_loss) + .with_max_usage_from_shore("MAX_USAGE_FROM_SHORE;MAX_USAGE_FROM_SHORE") + ).validate() + + installation_pfs = ( + YamlInstallationBuilder() + .with_name("minimal_installation") + .with_category(InstallationUserDefinedCategoryType.FIXED) + .with_fuel(fuel.name) + .with_generator_sets([generator_set]) + .with_regularity(regularity) + ).validate() + + resources = { + ltp_test_helper.generator_fuel_energy_function.name: generator_electricity2fuel_17MW_resource, + ltp_test_helper.power_from_shore_energy_function.name: onshore_power_electricity2fuel_resource, + max_usage_from_shore_time_series.name: max_usage_from_shore_time_series_resource, + } + + asset_pfs = ( + YamlAssetBuilder() + .with_start(time_vector_yearly[0]) + .with_end(time_vector_yearly[-1]) + .with_installations([installation_pfs]) + .with_fuel_types([fuel]) + .with_time_series([max_usage_from_shore_time_series]) + .with_facility_inputs( + [ltp_test_helper.generator_fuel_energy_function, ltp_test_helper.power_from_shore_energy_function] + ) + ).validate() + + asset_pfs = ltp_test_helper.get_yaml_model( + request, asset=asset_pfs, resources=resources, frequency=Frequency.YEAR + ) + ltp_result = ltp_test_helper.get_ltp_result(asset_pfs, asset_pfs.variables) + max_usage_from_shore = ltp_test_helper.get_ltp_column( + ltp_result=ltp_result, installation_nr=0, ltp_column="fromShorePeakMaximum" + ) + max_usage_from_shore_2027 = float( + max_usage_from_shore.values[Period(datetime(2027, 1, 1), datetime(2028, 1, 1))] + ) + # In the input memory resource max usage from shore is 250 (1.12.2026), 290 (1.6.2027), 283 (1.1.2028) + # and 283 (1.1.2029). Ensure that the correct value is set for 2027 (290 from 1.6): + assert max_usage_from_shore_2027 == 290.0 + + # Ensure that values in 2027, 2028 and 2029 are correct, based on input file: + assert [float(max_pfs) for max_pfs in max_usage_from_shore.values.values()][2:5] == [ + 290, + 283, + 283, + ]