From 6b0c72875e667ca1abce5b9b1f2ef4a9548d0d1e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ole=20Petter=20L=C3=B8d=C3=B8en?= Date: Fri, 27 Oct 2023 12:35:20 +0200 Subject: [PATCH] refactor: move validate operational conditions from compressor train, rename to validate model input (#256) chore: add ModelInputFailureStatus chore: fix tests --- src/libecalc/core/models/__init__.py | 1 + src/libecalc/core/models/compressor/base.py | 200 +---------------- .../core/models/compressor/train/base.py | 5 +- ...on_shaft_multiple_streams_and_pressures.py | 5 +- .../core/models/model_input_validation.py | 209 ++++++++++++++++++ .../test_compressor_train_base.py | 175 --------------- .../core/models/test_validate_model_input.py | 186 ++++++++++++++++ 7 files changed, 403 insertions(+), 378 deletions(-) create mode 100644 src/libecalc/core/models/model_input_validation.py create mode 100644 src/tests/libecalc/core/models/test_validate_model_input.py diff --git a/src/libecalc/core/models/__init__.py b/src/libecalc/core/models/__init__.py index e69de29bb2..c80982ff30 100644 --- a/src/libecalc/core/models/__init__.py +++ b/src/libecalc/core/models/__init__.py @@ -0,0 +1 @@ +from .model_input_validation import ModelInputFailureStatus, validate_model_input diff --git a/src/libecalc/core/models/compressor/base.py b/src/libecalc/core/models/compressor/base.py index 98ef17d163..e38ee9c92d 100644 --- a/src/libecalc/core/models/compressor/base.py +++ b/src/libecalc/core/models/compressor/base.py @@ -2,7 +2,7 @@ from abc import abstractmethod from functools import partial -from typing import List, Optional, Tuple +from typing import List import numpy as np from libecalc import dto @@ -15,9 +15,6 @@ ) from libecalc.core.models.compressor.train.utils.numeric_methods import find_root from libecalc.core.models.results import CompressorTrainResult -from libecalc.core.models.results.compressor import ( - CompressorTrainCommonShaftFailureStatus, -) from libecalc.core.models.turbine import TurbineModel from numpy.typing import NDArray @@ -71,201 +68,6 @@ def evaluate_streams( ): raise NotImplementedError - def validate_operational_conditions( - self, - rate: NDArray[np.float64], - suction_pressure: NDArray[np.float64], - discharge_pressure: NDArray[np.float64], - intermediate_pressure: Optional[NDArray[np.float64]] = None, - ) -> Tuple[ - NDArray[np.float64], - NDArray[np.float64], - NDArray[np.float64], - NDArray[np.float64], - List[CompressorTrainCommonShaftFailureStatus], - ]: - indices_to_validate = self._find_indices_to_validate(rate=rate) - validated_failure_status = [None] * len(suction_pressure) - validated_rate = rate.copy() - validated_suction_pressure = suction_pressure.copy() - validated_discharge_pressure = discharge_pressure.copy() - if intermediate_pressure is not None: - validated_intermediate_pressure = intermediate_pressure - if len(indices_to_validate) >= 1: - ( - tmp_rate, - tmp_suction_pressure, - tmp_discharge_pressure, - tmp_intermediate_pressure, - tmp_failure_status, - ) = self._validate_operational_conditions( - rate=rate[:, indices_to_validate] if np.ndim(rate) == 2 else rate[indices_to_validate], - suction_pressure=suction_pressure[indices_to_validate], - discharge_pressure=discharge_pressure[indices_to_validate], - intermediate_pressure=intermediate_pressure[indices_to_validate] - if intermediate_pressure is not None - else None, - ) - - if np.ndim(rate) == 2: - validated_rate[:, indices_to_validate] = tmp_rate - else: - validated_rate[indices_to_validate] = tmp_rate - validated_suction_pressure[indices_to_validate] = tmp_suction_pressure - validated_discharge_pressure[indices_to_validate] = tmp_discharge_pressure - if intermediate_pressure is not None: - validated_intermediate_pressure[indices_to_validate] = tmp_intermediate_pressure - for i, failure in enumerate(tmp_failure_status): - validated_failure_status[indices_to_validate[i]] = failure - - # any remaining zero or negative suction/discharge pressures (for unvalidated time steps, others are already changed) - # must be set to 1 (for neqsim to initiate fluid streams) - validated_suction_pressure = np.where(validated_suction_pressure <= 0, 1, validated_suction_pressure) - validated_discharge_pressure = np.where(validated_discharge_pressure <= 0, 1, validated_discharge_pressure) - - return ( - validated_rate, - validated_suction_pressure, - validated_discharge_pressure, - validated_intermediate_pressure if intermediate_pressure is not None else None, - validated_failure_status, - ) - - @staticmethod - def _find_indices_to_validate(rate: NDArray[np.float64]) -> List[int]: - """Find indices of array where rate(s) are positive. - For a 1D array, this means returning the indices where rate is positive. - For a 2D array, this means returning the indices where at least one rate is positive (along 0-axis). - """ - return np.where(np.any(rate != 0, axis=0) if np.ndim(rate) == 2 else rate != 0)[0].tolist() - - @staticmethod - def _validate_operational_conditions( - rate: NDArray[np.float64], - suction_pressure: NDArray[np.float64], - discharge_pressure: NDArray[np.float64], - intermediate_pressure: Optional[NDArray[np.float64]] = None, - ) -> Tuple[ - NDArray[np.float64], - NDArray[np.float64], - NDArray[np.float64], - NDArray[np.float64], - List[CompressorTrainCommonShaftFailureStatus], - ]: - """ - Checks for negative or zero values in the input values to the compressor train. - - The following is done: - - Time steps where rate is zero are not checked for validity - (but zero or negative pressures will still be changed to 1) - - Any pressures that are negative or zero are set to one, and all rates for that time step are set to zero - - Any negative rates are set to zero - - A failure_status describing the first failure encountered is returned - - Returns only one failure_status. Checks the potential failures at each time step in the following order: - suction pressure, intermediate_pressure, discharge pressure and rate. If there are multiple failures, - only the first one will be returned. When the input is changed to fix the first failure, the next failure - will be reported, and so on. - - Args: - rate: Input rate(s) to the compressor train - suction_pressure: Suction pressures for the compressor train - discharge_pressure: Discharge pressures for the compressor train - intermediate_pressure: Intermediate pressures for the compressor train (if any) - - Returns: - Tuple with the (potentially) updated input arrays and a failure_status describing if any input is invalid - - """ - validation_failures = [ - CompressorTrainCommonShaftFailureStatus.INVALID_SUCTION_PRESSURE_INPUT, - CompressorTrainCommonShaftFailureStatus.INVALID_INTERMEDIATE_PRESSURE_INPUT, - CompressorTrainCommonShaftFailureStatus.INVALID_DISCHARGE_PRESSURE_INPUT, - CompressorTrainCommonShaftFailureStatus.INVALID_RATE_INPUT, - None, - ] - - input_rate = rate.copy() - input_suction_pressure = suction_pressure.copy() - input_intermediate_pressure = intermediate_pressure.copy() if intermediate_pressure is not None else None - input_discharge_pressure = discharge_pressure.copy() - - if not np.all(rate >= 0): - logger.warning( - f"The rate(s) in the compressor train must have non negative values. Given values [Sm3/sd]: " - f"{rate.tolist()}. The affected time steps will not be calculated, and rate is set to zero." - ) - rate = ( - np.where(np.any(rate < 0, axis=0), 0, rate) if np.ndim(rate) == 2 else np.where(rate < 0, 0, rate) - ) # if the rate for one stream is negative, set the rates for all streams to zero for that time step - if intermediate_pressure is not None: - if not np.all(intermediate_pressure > 0): - logger.warning( - f"Interstage pressure needs to be a positive value. Given values: {intermediate_pressure.tolist()}." - f" The affected time steps will not be calculated, and rate is set to zero." - ) - rate = np.where(intermediate_pressure <= 0, 0, rate) - intermediate_pressure = np.where(intermediate_pressure <= 0, 1, intermediate_pressure) - if not np.all(suction_pressure > 0): - logger.warning( - f"Inlet pressure needs to be a positive value. Given values: {suction_pressure.tolist()}." - f" The affected time steps will not be calculated, and rate is set to zero." - ) - rate = np.where(suction_pressure <= 0, 0, rate) - suction_pressure = np.where(suction_pressure <= 0, 1, suction_pressure) - if not np.all(discharge_pressure > 0): - logger.warning( - f"Outlet pressure needs to be a positive value. Given values: {discharge_pressure.tolist()}" - f" The affected time steps will not be calculated, and rate is set to zero." - ) - rate = np.where(discharge_pressure <= 0, 0, rate) - discharge_pressure = np.where(discharge_pressure <= 0, 1, discharge_pressure) - if not np.all(discharge_pressure >= suction_pressure): - logger.warning( - f"Inlet pressure needs to be a less than or equal to outlet pressure. Given values for inlet" - f" pressure: {suction_pressure.tolist()}. Given values for outlet pressure:" - f" {discharge_pressure.tolist()}. The affected time steps will not be calculated," - f" and rate is set to zero." - ) - rate = np.where(discharge_pressure < suction_pressure, 0, rate) - # for multiple stream train, rate is 2D - if np.ndim(rate) == 2: - # check if any of the streams have changed value during validation, streams along axis 0, time along axis 1 - invalid_rate_input = np.any(rate != input_rate, axis=0) - else: - invalid_rate_input = np.where(rate != input_rate, True, False) - - invalid_suction_pressure_input = np.logical_or( - np.where(suction_pressure != input_suction_pressure, True, False), - np.where(suction_pressure > discharge_pressure, True, False), - ) - invalid_discharge_pressure_input = np.where(discharge_pressure != input_discharge_pressure, True, False) - invalid_intermediate_pressure_input = ( - np.where(intermediate_pressure != input_intermediate_pressure, True, False) - if intermediate_pressure is not None - else np.asarray([False] * len(suction_pressure)) - ) - - failure_status = [ - validation_failures[ - [ - invalid_suction_pressure, - invalid_intermediate_pressure, - invalid_discharge_pressure, - invalid_rate, - True, # This is to also pick up failure_status None - ].index(True) - ] - for invalid_rate, invalid_suction_pressure, invalid_intermediate_pressure, invalid_discharge_pressure in zip( - invalid_rate_input, - invalid_suction_pressure_input, - invalid_intermediate_pressure_input, - invalid_discharge_pressure_input, - ) - ] - - return rate, suction_pressure, discharge_pressure, intermediate_pressure, failure_status - class CompressorWithTurbineModel(CompressorModel): def __init__( diff --git a/src/libecalc/core/models/compressor/train/base.py b/src/libecalc/core/models/compressor/train/base.py index 2d625c17d1..2aa21ebe18 100644 --- a/src/libecalc/core/models/compressor/train/base.py +++ b/src/libecalc/core/models/compressor/train/base.py @@ -7,6 +7,7 @@ from libecalc.common.logger import logger from libecalc.common.stream import Stream from libecalc.common.units import Unit +from libecalc.core.models import ModelInputFailureStatus, validate_model_input from libecalc.core.models.compressor.base import CompressorModel from libecalc.core.models.compressor.results import CompressorTrainResultSingleTimeStep from libecalc.core.models.compressor.train.fluid import FluidStream @@ -84,7 +85,7 @@ def evaluate_rate_ps_pd( """ logger.debug(f"Evaluating {type(self).__name__} given rate, suction and discharge pressure.") - rate, suction_pressure, discharge_pressure, _, input_failure_status = self.validate_operational_conditions( + rate, suction_pressure, discharge_pressure, _, input_failure_status = validate_model_input( rate=rate, suction_pressure=suction_pressure, discharge_pressure=discharge_pressure, @@ -125,7 +126,7 @@ def evaluate_rate_ps_pd( ) for i, train_result in enumerate(train_results): - if input_failure_status[i]: + if input_failure_status[i] is not ModelInputFailureStatus.NO_FAILURE: train_result.failure_status = input_failure_status[i] return CompressorTrainResult( diff --git a/src/libecalc/core/models/compressor/train/variable_speed_compressor_train_common_shaft_multiple_streams_and_pressures.py b/src/libecalc/core/models/compressor/train/variable_speed_compressor_train_common_shaft_multiple_streams_and_pressures.py index de7232d572..da1fb2e4ce 100644 --- a/src/libecalc/core/models/compressor/train/variable_speed_compressor_train_common_shaft_multiple_streams_and_pressures.py +++ b/src/libecalc/core/models/compressor/train/variable_speed_compressor_train_common_shaft_multiple_streams_and_pressures.py @@ -8,6 +8,7 @@ from libecalc.common.logger import logger from libecalc.common.stream import Stream from libecalc.common.units import Unit, UnitConstants +from libecalc.core.models import ModelInputFailureStatus, validate_model_input from libecalc.core.models.compressor.results import CompressorTrainResultSingleTimeStep from libecalc.core.models.compressor.train.base import CompressorTrainModel from libecalc.core.models.compressor.train.fluid import FluidStream @@ -774,7 +775,7 @@ def evaluate_rate_ps_pint_pd( discharge_pressure, intermediate_pressure, input_failure_status, - ) = self.validate_operational_conditions( + ) = validate_model_input( rate=rate, suction_pressure=suction_pressure, discharge_pressure=discharge_pressure, @@ -825,7 +826,7 @@ def evaluate_rate_ps_pint_pd( ) for i, train_result in enumerate(train_results): - if input_failure_status[i]: + if input_failure_status[i] is not ModelInputFailureStatus.NO_FAILURE: train_result.failure_status = input_failure_status[i] return CompressorTrainResult( diff --git a/src/libecalc/core/models/model_input_validation.py b/src/libecalc/core/models/model_input_validation.py new file mode 100644 index 0000000000..8d816e2053 --- /dev/null +++ b/src/libecalc/core/models/model_input_validation.py @@ -0,0 +1,209 @@ +from enum import Enum +from typing import List, Optional, Tuple + +import numpy as np +from libecalc.common.logger import logger +from numpy.typing import NDArray + + +class ModelInputFailureStatus(str, Enum): + INVALID_RATE_INPUT = "INVALID_RATE_INPUT" + INVALID_SUCTION_PRESSURE_INPUT = "INVALID_SUCTION_PRESSURE_INPUT" + INVALID_INTERMEDIATE_PRESSURE_INPUT = "INVALID_INTERMEDIATE_PRESSURE_INPUT" + INVALID_DISCHARGE_PRESSURE_INPUT = "INVALID_DISCHARGE_PRESSURE_INPUT" + NO_FAILURE = None + + +def validate_model_input( + rate: NDArray[np.float64], + suction_pressure: NDArray[np.float64], + discharge_pressure: NDArray[np.float64], + intermediate_pressure: Optional[NDArray[np.float64]] = None, +) -> Tuple[ + NDArray[np.float64], + NDArray[np.float64], + NDArray[np.float64], + NDArray[np.float64], + List[ModelInputFailureStatus], +]: + indices_to_validate = _find_indices_to_validate(rate=rate) + validated_failure_status = [None] * len(suction_pressure) + validated_rate = rate.copy() + validated_suction_pressure = suction_pressure.copy() + validated_discharge_pressure = discharge_pressure.copy() + if intermediate_pressure is not None: + validated_intermediate_pressure = intermediate_pressure + if len(indices_to_validate) >= 1: + ( + tmp_rate, + tmp_suction_pressure, + tmp_discharge_pressure, + tmp_intermediate_pressure, + tmp_failure_status, + ) = _validate_model_input( + rate=rate[:, indices_to_validate] if np.ndim(rate) == 2 else rate[indices_to_validate], + suction_pressure=suction_pressure[indices_to_validate], + discharge_pressure=discharge_pressure[indices_to_validate], + intermediate_pressure=intermediate_pressure[indices_to_validate] + if intermediate_pressure is not None + else None, + ) + + if np.ndim(rate) == 2: + validated_rate[:, indices_to_validate] = tmp_rate + else: + validated_rate[indices_to_validate] = tmp_rate + validated_suction_pressure[indices_to_validate] = tmp_suction_pressure + validated_discharge_pressure[indices_to_validate] = tmp_discharge_pressure + if intermediate_pressure is not None: + validated_intermediate_pressure[indices_to_validate] = tmp_intermediate_pressure + for i, failure in enumerate(tmp_failure_status): + validated_failure_status[indices_to_validate[i]] = failure + + # any remaining zero or negative suction/discharge pressures (for unvalidated time steps, others are already changed) + # must be set to 1 (for neqsim to initiate fluid streams) + validated_suction_pressure = np.where(validated_suction_pressure <= 0, 1, validated_suction_pressure) + validated_discharge_pressure = np.where(validated_discharge_pressure <= 0, 1, validated_discharge_pressure) + + return ( + validated_rate, + validated_suction_pressure, + validated_discharge_pressure, + validated_intermediate_pressure if intermediate_pressure is not None else None, + validated_failure_status, + ) + + +def _find_indices_to_validate(rate: NDArray[np.float64]) -> List[int]: + """Find indices of array where rate(s) are positive. + For a 1D array, this means returning the indices where rate is positive. + For a 2D array, this means returning the indices where at least one rate is positive (along 0-axis). + """ + return np.where(np.any(rate != 0, axis=0) if np.ndim(rate) == 2 else rate != 0)[0].tolist() + + +def _validate_model_input( + rate: NDArray[np.float64], + suction_pressure: NDArray[np.float64], + discharge_pressure: NDArray[np.float64], + intermediate_pressure: Optional[NDArray[np.float64]] = None, +) -> Tuple[ + NDArray[np.float64], + NDArray[np.float64], + NDArray[np.float64], + NDArray[np.float64], + List[ModelInputFailureStatus], +]: + """ + Checks for negative or zero values in the input values to the compressor train. + + The following is done: + - Time steps where rate is zero are not checked for validity + (but zero or negative pressures will still be changed to 1) + - Any pressures that are negative or zero are set to one, and all rates for that time step are set to zero + - Any negative rates are set to zero + - A failure_status describing the first failure encountered is returned + + Returns only one failure_status. Checks the potential failures at each time step in the following order: + suction pressure, intermediate_pressure, discharge pressure and rate. If there are multiple failures, + only the first one will be returned. When the input is changed to fix the first failure, the next failure + will be reported, and so on. + + Args: + rate: Input rate(s) to the compressor train + suction_pressure: Suction pressures for the compressor train + discharge_pressure: Discharge pressures for the compressor train + intermediate_pressure: Intermediate pressures for the compressor train (if any) + + Returns: + Tuple with the (potentially) updated input arrays and a failure_status describing if any input is invalid + + """ + validation_failures = [ + ModelInputFailureStatus.INVALID_SUCTION_PRESSURE_INPUT, + ModelInputFailureStatus.INVALID_INTERMEDIATE_PRESSURE_INPUT, + ModelInputFailureStatus.INVALID_DISCHARGE_PRESSURE_INPUT, + ModelInputFailureStatus.INVALID_RATE_INPUT, + ModelInputFailureStatus.NO_FAILURE, + ] + + input_rate = rate.copy() + input_suction_pressure = suction_pressure.copy() + input_intermediate_pressure = intermediate_pressure.copy() if intermediate_pressure is not None else None + input_discharge_pressure = discharge_pressure.copy() + + if not np.all(rate >= 0): + logger.warning( + f"The rate(s) in the compressor train must have non negative values. Given values [Sm3/sd]: " + f"{rate.tolist()}. The affected time steps will not be calculated, and rate is set to zero." + ) + rate = ( + np.where(np.any(rate < 0, axis=0), 0, rate) if np.ndim(rate) == 2 else np.where(rate < 0, 0, rate) + ) # if the rate for one stream is negative, set the rates for all streams to zero for that time step + if intermediate_pressure is not None: + if not np.all(intermediate_pressure > 0): + logger.warning( + f"Interstage pressure needs to be a positive value. Given values: {intermediate_pressure.tolist()}." + f" The affected time steps will not be calculated, and rate is set to zero." + ) + rate = np.where(intermediate_pressure <= 0, 0, rate) + intermediate_pressure = np.where(intermediate_pressure <= 0, 1, intermediate_pressure) + if not np.all(suction_pressure > 0): + logger.warning( + f"Inlet pressure needs to be a positive value. Given values: {suction_pressure.tolist()}." + f" The affected time steps will not be calculated, and rate is set to zero." + ) + rate = np.where(suction_pressure <= 0, 0, rate) + suction_pressure = np.where(suction_pressure <= 0, 1, suction_pressure) + if not np.all(discharge_pressure > 0): + logger.warning( + f"Outlet pressure needs to be a positive value. Given values: {discharge_pressure.tolist()}" + f" The affected time steps will not be calculated, and rate is set to zero." + ) + rate = np.where(discharge_pressure <= 0, 0, rate) + discharge_pressure = np.where(discharge_pressure <= 0, 1, discharge_pressure) + if not np.all(discharge_pressure >= suction_pressure): + logger.warning( + f"Inlet pressure needs to be a less than or equal to outlet pressure. Given values for inlet" + f" pressure: {suction_pressure.tolist()}. Given values for outlet pressure:" + f" {discharge_pressure.tolist()}. The affected time steps will not be calculated," + f" and rate is set to zero." + ) + rate = np.where(discharge_pressure < suction_pressure, 0, rate) + # for multiple stream train, rate is 2D + if np.ndim(rate) == 2: + # check if any of the streams have changed value during validation, streams along axis 0, time along axis 1 + invalid_rate_input = np.any(rate != input_rate, axis=0) + else: + invalid_rate_input = np.where(rate != input_rate, True, False) + + invalid_suction_pressure_input = np.logical_or( + np.where(suction_pressure != input_suction_pressure, True, False), + np.where(suction_pressure > discharge_pressure, True, False), + ) + invalid_discharge_pressure_input = np.where(discharge_pressure != input_discharge_pressure, True, False) + invalid_intermediate_pressure_input = ( + np.where(intermediate_pressure != input_intermediate_pressure, True, False) + if intermediate_pressure is not None + else np.asarray([False] * len(suction_pressure)) + ) + + failure_status = [ + validation_failures[ + [ + invalid_suction_pressure, + invalid_intermediate_pressure, + invalid_discharge_pressure, + invalid_rate, + True, # This is to also pick up failure_status NO_FAILURE + ].index(True) + ] + for invalid_rate, invalid_suction_pressure, invalid_intermediate_pressure, invalid_discharge_pressure in zip( + invalid_rate_input, + invalid_suction_pressure_input, + invalid_intermediate_pressure_input, + invalid_discharge_pressure_input, + ) + ] + + return rate, suction_pressure, discharge_pressure, intermediate_pressure, failure_status diff --git a/src/tests/libecalc/core/models/compressor_modelling/test_compressor_train_base.py b/src/tests/libecalc/core/models/compressor_modelling/test_compressor_train_base.py index 94b47772a9..553c69f865 100644 --- a/src/tests/libecalc/core/models/compressor_modelling/test_compressor_train_base.py +++ b/src/tests/libecalc/core/models/compressor_modelling/test_compressor_train_base.py @@ -4,9 +4,6 @@ import pytest from libecalc import dto from libecalc.core.models.compressor.train.base import CompressorTrainModel -from libecalc.core.models.results.compressor import ( - CompressorTrainCommonShaftFailureStatus, -) @pytest.fixture @@ -40,175 +37,3 @@ def test_calculate_pressure_ratios_per_stage(compressor_train_two_stages): ) np.testing.assert_almost_equal(pressure_ratios, [1.0, 1.22474487, 1.41421356, 1.58113883, 1.73205081]) - - -def test_validate_operational_conditions(compressor_train): - # test that valid single stream input is not changed - [ - rate, - suction_pressure, - discharge_pressure, - intermediate_pressure, - failure_status, - ] = compressor_train.validate_operational_conditions( - rate=np.asarray([1000, 1000]), - suction_pressure=np.asarray([1, 1]), - intermediate_pressure=np.asarray([2, 2]), - discharge_pressure=np.asarray([3, 3]), - ) - assert np.all(rate == 1000) - assert np.all(suction_pressure == 1) - assert np.all(intermediate_pressure == 2) - assert np.all(discharge_pressure == 3) - assert failure_status == [None, None] - - # test that valid multiple stream input is not changed - [ - rate, - suction_pressure, - discharge_pressure, - intermediate_pressure, - failure_status, - ] = compressor_train.validate_operational_conditions( - rate=np.asarray([[1000, 1000], [1000, 1000]]), - suction_pressure=np.asarray([1, 1]), - intermediate_pressure=np.asarray([2, 2]), - discharge_pressure=np.asarray([3, 3]), - ) - assert np.all(rate == 1000) - assert np.all(suction_pressure == 1) - assert np.all(intermediate_pressure == 2) - assert np.all(discharge_pressure == 3) - assert failure_status == [None, None] - - # test that suction pressure <= 0 is set to 1 and that the corresponding rate is set to 0 - [ - rate, - suction_pressure, - discharge_pressure, - intermediate_pressure, - failure_status, - ] = compressor_train.validate_operational_conditions( - rate=np.asarray([1000, 1000]), - suction_pressure=np.asarray([-1, 0]), - intermediate_pressure=np.asarray([2, 2]), - discharge_pressure=np.asarray([3, 3]), - ) - assert np.all(rate == 0) - assert np.all(suction_pressure == 1) - assert np.all(intermediate_pressure == 2) - assert np.all(discharge_pressure == 3) - assert failure_status == [ - CompressorTrainCommonShaftFailureStatus.INVALID_SUCTION_PRESSURE_INPUT, - CompressorTrainCommonShaftFailureStatus.INVALID_SUCTION_PRESSURE_INPUT, - ] - - # test that discharge pressure <= 0 is set to 1 and that the corresponding rate is set to 0 - [ - rate, - suction_pressure, - discharge_pressure, - intermediate_pressure, - failure_status, - ] = compressor_train.validate_operational_conditions( - rate=np.asarray([1000, 1000]), - suction_pressure=np.asarray([1, 1]), - intermediate_pressure=np.asarray([2, 2]), - discharge_pressure=np.asarray([-1, 0]), - ) - assert np.all(rate == 0) - assert np.all(suction_pressure == 1) - assert np.all(intermediate_pressure == 2) - assert np.all(discharge_pressure == 1) - assert failure_status == [ - CompressorTrainCommonShaftFailureStatus.INVALID_DISCHARGE_PRESSURE_INPUT, - CompressorTrainCommonShaftFailureStatus.INVALID_DISCHARGE_PRESSURE_INPUT, - ] - - # test that intermediate prressure <= 0 is set to 1 and that the corresponding rate is set to 0 - [ - rate, - suction_pressure, - discharge_pressure, - intermediate_pressure, - failure_status, - ] = compressor_train.validate_operational_conditions( - rate=np.asarray([1000, 1000]), - suction_pressure=np.asarray([1, 1]), - intermediate_pressure=np.asarray([-1, 0]), - discharge_pressure=np.asarray([3, 3]), - ) - assert np.all(rate == 0) - assert np.all(suction_pressure == 1) - assert np.all(intermediate_pressure == 1) - assert np.all(discharge_pressure == 3) - assert failure_status == [ - CompressorTrainCommonShaftFailureStatus.INVALID_INTERMEDIATE_PRESSURE_INPUT, - CompressorTrainCommonShaftFailureStatus.INVALID_INTERMEDIATE_PRESSURE_INPUT, - ] - - # test that rate < 0 is set to 0 for multiple stream compressor train - [ - rate, - suction_pressure, - discharge_pressure, - intermediate_pressure, - failure_status, - ] = compressor_train.validate_operational_conditions( - rate=np.asarray([[-1, 1000, 1000], [1000, -1, 1000]]), - suction_pressure=np.asarray([1, 1, 1]), - intermediate_pressure=np.asarray([2, 2, 2]), - discharge_pressure=np.asarray([3, 3, 3]), - ) - assert np.all(rate[:, 0] == 0) - assert np.all(rate[:, 1] == 0) - assert np.all(rate[:, 2] == 1000) - assert np.all(suction_pressure == 1) - assert np.all(intermediate_pressure == 2) - assert np.all(discharge_pressure == 3) - assert failure_status == [ - CompressorTrainCommonShaftFailureStatus.INVALID_RATE_INPUT, - CompressorTrainCommonShaftFailureStatus.INVALID_RATE_INPUT, - None, - ] - # test that rate < 0 is set to 0 for single stream compressor train - [ - rate, - suction_pressure, - discharge_pressure, - intermediate_pressure, - failure_status, - ] = compressor_train.validate_operational_conditions( - rate=np.asarray([-1, 1000]), - suction_pressure=np.asarray([1, 1]), - intermediate_pressure=np.asarray([2, 2]), - discharge_pressure=np.asarray([3, 3]), - ) - assert rate[0] == 0 - assert rate[1] == 1000 - assert np.all(suction_pressure == 1) - assert np.all(intermediate_pressure == 2) - assert np.all(discharge_pressure == 3) - assert failure_status == [CompressorTrainCommonShaftFailureStatus.INVALID_RATE_INPUT, None] - - # test that suction_pressure, when suction_pressure > discharge_pressure, is set to discharge_pressure, - # and that rate at the same time is set to 0. - [ - rate, - suction_pressure, - discharge_pressure, - intermediate_pressure, - failure_status, - ] = compressor_train.validate_operational_conditions( - rate=np.asarray([1000, 1000]), - suction_pressure=np.asarray([4, 3]), - intermediate_pressure=np.asarray([2, 2]), - discharge_pressure=np.asarray([3, 3]), - ) - assert rate[0] == 0 - assert rate[1] == 1000 - assert suction_pressure[0] == 4 - assert suction_pressure[1] == 3 - assert np.all(intermediate_pressure == 2) - assert np.all(discharge_pressure == 3) - assert failure_status == [CompressorTrainCommonShaftFailureStatus.INVALID_SUCTION_PRESSURE_INPUT, None] diff --git a/src/tests/libecalc/core/models/test_validate_model_input.py b/src/tests/libecalc/core/models/test_validate_model_input.py new file mode 100644 index 0000000000..e6dec733a0 --- /dev/null +++ b/src/tests/libecalc/core/models/test_validate_model_input.py @@ -0,0 +1,186 @@ +import numpy as np +from libecalc.core.models import ModelInputFailureStatus, validate_model_input + + +def test_validate_model_input(): + # test that valid single stream input is not changed + [ + rate, + suction_pressure, + discharge_pressure, + intermediate_pressure, + failure_status, + ] = validate_model_input( + rate=np.asarray([1000, 1000]), + suction_pressure=np.asarray([1, 1]), + intermediate_pressure=np.asarray([2, 2]), + discharge_pressure=np.asarray([3, 3]), + ) + assert np.all(rate == 1000) + assert np.all(suction_pressure == 1) + assert np.all(intermediate_pressure == 2) + assert np.all(discharge_pressure == 3) + assert failure_status == [ + ModelInputFailureStatus.NO_FAILURE, + ModelInputFailureStatus.NO_FAILURE, + ] + + # test that valid multiple stream input is not changed + [ + rate, + suction_pressure, + discharge_pressure, + intermediate_pressure, + failure_status, + ] = validate_model_input( + rate=np.asarray([[1000, 1000], [1000, 1000]]), + suction_pressure=np.asarray([1, 1]), + intermediate_pressure=np.asarray([2, 2]), + discharge_pressure=np.asarray([3, 3]), + ) + assert np.all(rate == 1000) + assert np.all(suction_pressure == 1) + assert np.all(intermediate_pressure == 2) + assert np.all(discharge_pressure == 3) + assert failure_status == [ + ModelInputFailureStatus.NO_FAILURE, + ModelInputFailureStatus.NO_FAILURE, + ] + + # test that suction pressure <= 0 is set to 1 and that the corresponding rate is set to 0 + [ + rate, + suction_pressure, + discharge_pressure, + intermediate_pressure, + failure_status, + ] = validate_model_input( + rate=np.asarray([1000, 1000]), + suction_pressure=np.asarray([-1, 0]), + intermediate_pressure=np.asarray([2, 2]), + discharge_pressure=np.asarray([3, 3]), + ) + assert np.all(rate == 0) + assert np.all(suction_pressure == 1) + assert np.all(intermediate_pressure == 2) + assert np.all(discharge_pressure == 3) + assert failure_status == [ + ModelInputFailureStatus.INVALID_SUCTION_PRESSURE_INPUT, + ModelInputFailureStatus.INVALID_SUCTION_PRESSURE_INPUT, + ] + + # test that discharge pressure <= 0 is set to 1 and that the corresponding rate is set to 0 + [ + rate, + suction_pressure, + discharge_pressure, + intermediate_pressure, + failure_status, + ] = validate_model_input( + rate=np.asarray([1000, 1000]), + suction_pressure=np.asarray([1, 1]), + intermediate_pressure=np.asarray([2, 2]), + discharge_pressure=np.asarray([-1, 0]), + ) + assert np.all(rate == 0) + assert np.all(suction_pressure == 1) + assert np.all(intermediate_pressure == 2) + assert np.all(discharge_pressure == 1) + assert failure_status == [ + ModelInputFailureStatus.INVALID_DISCHARGE_PRESSURE_INPUT, + ModelInputFailureStatus.INVALID_DISCHARGE_PRESSURE_INPUT, + ] + + # test that intermediate prressure <= 0 is set to 1 and that the corresponding rate is set to 0 + [ + rate, + suction_pressure, + discharge_pressure, + intermediate_pressure, + failure_status, + ] = validate_model_input( + rate=np.asarray([1000, 1000]), + suction_pressure=np.asarray([1, 1]), + intermediate_pressure=np.asarray([-1, 0]), + discharge_pressure=np.asarray([3, 3]), + ) + assert np.all(rate == 0) + assert np.all(suction_pressure == 1) + assert np.all(intermediate_pressure == 1) + assert np.all(discharge_pressure == 3) + assert failure_status == [ + ModelInputFailureStatus.INVALID_INTERMEDIATE_PRESSURE_INPUT, + ModelInputFailureStatus.INVALID_INTERMEDIATE_PRESSURE_INPUT, + ] + + # test that rate < 0 is set to 0 for multiple stream + [ + rate, + suction_pressure, + discharge_pressure, + intermediate_pressure, + failure_status, + ] = validate_model_input( + rate=np.asarray([[-1, 1000, 1000], [1000, -1, 1000]]), + suction_pressure=np.asarray([1, 1, 1]), + intermediate_pressure=np.asarray([2, 2, 2]), + discharge_pressure=np.asarray([3, 3, 3]), + ) + assert np.all(rate[:, 0] == 0) + assert np.all(rate[:, 1] == 0) + assert np.all(rate[:, 2] == 1000) + assert np.all(suction_pressure == 1) + assert np.all(intermediate_pressure == 2) + assert np.all(discharge_pressure == 3) + assert failure_status == [ + ModelInputFailureStatus.INVALID_RATE_INPUT, + ModelInputFailureStatus.INVALID_RATE_INPUT, + ModelInputFailureStatus.NO_FAILURE, + ] + # test that rate < 0 is set to 0 for single stream + [ + rate, + suction_pressure, + discharge_pressure, + intermediate_pressure, + failure_status, + ] = validate_model_input( + rate=np.asarray([-1, 1000]), + suction_pressure=np.asarray([1, 1]), + intermediate_pressure=np.asarray([2, 2]), + discharge_pressure=np.asarray([3, 3]), + ) + assert rate[0] == 0 + assert rate[1] == 1000 + assert np.all(suction_pressure == 1) + assert np.all(intermediate_pressure == 2) + assert np.all(discharge_pressure == 3) + assert failure_status == [ + ModelInputFailureStatus.INVALID_RATE_INPUT, + ModelInputFailureStatus.NO_FAILURE, + ] + + # test that suction_pressure, when suction_pressure > discharge_pressure, is set to discharge_pressure, + # and that rate at the same time is set to 0. + [ + rate, + suction_pressure, + discharge_pressure, + intermediate_pressure, + failure_status, + ] = validate_model_input( + rate=np.asarray([1000, 1000]), + suction_pressure=np.asarray([4, 3]), + intermediate_pressure=np.asarray([2, 2]), + discharge_pressure=np.asarray([3, 3]), + ) + assert rate[0] == 0 + assert rate[1] == 1000 + assert suction_pressure[0] == 4 + assert suction_pressure[1] == 3 + assert np.all(intermediate_pressure == 2) + assert np.all(discharge_pressure == 3) + assert failure_status == [ + ModelInputFailureStatus.INVALID_SUCTION_PRESSURE_INPUT, + ModelInputFailureStatus.NO_FAILURE, + ]