chore: remove pydantic validation from dto classes

src/libecalc/core/models/compressor/sampled/compressor_model_sampled.py

@@ -167,8 +167,15 @@ def evaluate_rate_ps_pd(
         """
         # subtract an epsilon to make robust comparison.
         if rate is not None:
+            # Ensure rate is a NumPy array
+            rate = np.array(rate, dtype=np.float64)
             # To avoid bringing rate below zero.
             rate = np.where(rate > 0, rate - EPSILON, rate)
+        if suction_pressure is not None:
+            suction_pressure = np.array(suction_pressure, dtype=np.float64)
+
+        if discharge_pressure is not None:
+            discharge_pressure = np.array(discharge_pressure, dtype=np.float64)
 
         number_of_data_points = 0
         if rate is not None:

src/libecalc/core/models/model_input_validation.py

@@ -29,12 +29,18 @@ def validate_model_input(
     NDArray[np.float64],
     list[ModelInputFailureStatus],
 ]:
+    # Ensure input is a NumPy array
+    rate = np.array(rate, dtype=np.float64)
+    suction_pressure = np.array(suction_pressure, dtype=np.float64)
+    discharge_pressure = np.array(discharge_pressure, dtype=np.float64)
+
     indices_to_validate = _find_indices_to_validate(rate=rate)
     validated_failure_status = [ModelInputFailureStatus.NO_FAILURE] * 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:
+        intermediate_pressure = np.array(intermediate_pressure, dtype=np.float64)
         validated_intermediate_pressure = intermediate_pressure
     if len(indices_to_validate) >= 1:
         (
@@ -82,6 +88,7 @@ def _find_indices_to_validate(rate: NDArray[np.float64]) -> list[int]:
     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).
     """
+    rate = np.atleast_1d(rate)  # Ensure rate is at least 1D
     return np.where(np.any(rate != 0, axis=0) if np.ndim(rate) == 2 else rate != 0)[0].tolist()
 
 

src/libecalc/core/models/pump/pump.py

@@ -63,9 +63,13 @@ def evaluate_streams(
 
     @staticmethod
     def _calculate_head(
-        ps: NDArray[np.float64], pd: NDArray[np.float64], density: Union[NDArray[np.float64], float]
+        ps: Union[NDArray[np.float64], list[float]],
+        pd: Union[NDArray[np.float64], list[float]],
+        density: Union[NDArray[np.float64], float],
     ) -> NDArray[np.float64]:
         """:return: Head in joule per kg [J/kg]"""
+        ps = np.array(ps, dtype=np.float64)
+        pd = np.array(pd, dtype=np.float64)
         return np.array(Unit.BARA.to(Unit.PASCAL)(pd - ps) / density)
 
     @staticmethod
@@ -288,6 +292,10 @@ def evaluate_rate_ps_pd_density(
         :param discharge_pressures:
         :param fluid_density:
         """
+        # Ensure rate is a NumPy array
+        rate = np.asarray(rate, dtype=np.float64)
+        fluid_density = np.asarray(fluid_density, dtype=np.float64)
+
         # Ensure that the pump does not run when rate is <= 0.
         stream_day_rate = np.where(rate > 0, rate, 0)
 

src/libecalc/domain/infrastructure/__init__.py

@@ -1,5 +1,4 @@
 from libecalc.domain.infrastructure.energy_components.asset.asset import Asset
-from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseConsumer
 from libecalc.domain.infrastructure.energy_components.electricity_consumer.electricity_consumer import (
     ElectricityConsumer,
 )

src/libecalc/domain/infrastructure/energy_components/asset/asset.py

@@ -1,26 +1,24 @@
-from typing import Literal
-
-from pydantic import Field
-
 from libecalc.application.energy.energy_component import EnergyComponent
 from libecalc.common.component_type import ComponentType
 from libecalc.common.string.string_utils import generate_id
-from libecalc.domain.infrastructure.energy_components.base.component_dto import Component
 from libecalc.domain.infrastructure.energy_components.installation.installation import Installation
 from libecalc.dto.component_graph import ComponentGraph
-from libecalc.dto.utils.validators import ComponentNameStr
 
 
-class Asset(Component, EnergyComponent):
+class Asset(EnergyComponent):
+    def __init__(
+        self,
+        name: str,
+        installations: list[Installation],
+    ):
+        self.name = name
+        self.installations = installations
+        self.component_type = ComponentType.ASSET
+
     @property
     def id(self):
         return generate_id(self.name)
 
-    name: ComponentNameStr
-
-    installations: list[Installation] = Field(default_factory=list)
-    component_type: Literal[ComponentType.ASSET] = ComponentType.ASSET
-
     def is_fuel_consumer(self) -> bool:
         return True
 

src/libecalc/domain/infrastructure/energy_components/base/component_dto.py

@@ -1,106 +1,31 @@
-from abc import ABC, abstractmethod
-from typing import Optional
+from __future__ import annotations
 
-from pydantic import ConfigDict, Field, field_validator
-from pydantic_core.core_schema import ValidationInfo
+from typing import Optional
 
-from libecalc.common.component_type import ComponentType
-from libecalc.common.consumption_type import ConsumptionType
-from libecalc.common.string.string_utils import generate_id
-from libecalc.common.time_utils import Period
 from libecalc.common.units import Unit
 from libecalc.common.utils.rates import RateType
-from libecalc.domain.infrastructure.energy_components.utils import _convert_keys_in_dictionary_from_str_to_periods
-from libecalc.dto.base import EcalcBaseModel
-from libecalc.dto.fuel_type import FuelType
-from libecalc.dto.types import ConsumerUserDefinedCategoryType
-from libecalc.dto.utils.validators import (
-    ComponentNameStr,
-    ExpressionType,
-    validate_temporal_model,
-)
-from libecalc.expression import Expression
-
-
-class Component(EcalcBaseModel, ABC):
-    component_type: ComponentType
-
-    @property
-    @abstractmethod
-    def id(self) -> str: ...
-
-
-class BaseComponent(Component, ABC):
-    name: ComponentNameStr
-
-    regularity: dict[Period, Expression]
-
-    _validate_base_temporal_model = field_validator("regularity")(validate_temporal_model)
-
-    @field_validator("regularity", mode="before")
-    @classmethod
-    def check_regularity(cls, regularity):
-        if isinstance(regularity, dict) and len(regularity.values()) > 0:
-            regularity = _convert_keys_in_dictionary_from_str_to_periods(regularity)
-        return regularity
-
+from libecalc.dto.utils.validators import ExpressionType
 
-class BaseEquipment(BaseComponent, ABC):
-    user_defined_category: dict[Period, ConsumerUserDefinedCategoryType] = Field(..., validate_default=True)
 
-    @property
-    def id(self) -> str:
-        return generate_id(self.name)
+class ExpressionTimeSeries:
+    def __init__(self, value: ExpressionType, unit: Unit, type: Optional[RateType] = None):
+        self.value = value
+        self.unit = unit
+        self.type = type
 
-    @field_validator("user_defined_category", mode="before")
-    def check_user_defined_category(cls, user_defined_category, info: ValidationInfo):
-        """Provide which value and context to make it easier for user to correct wrt mandatory changes."""
-        if isinstance(user_defined_category, dict) and len(user_defined_category.values()) > 0:
-            user_defined_category = _convert_keys_in_dictionary_from_str_to_periods(user_defined_category)
-            for user_category in user_defined_category.values():
-                if user_category not in list(ConsumerUserDefinedCategoryType):
-                    name_context_str = ""
-                    if (name := info.data.get("name")) is not None:
-                        name_context_str = f"with the name {name}"
 
-                    raise ValueError(
-                        f"CATEGORY: {user_category} is not allowed for {cls.__name__} {name_context_str}. Valid categories are: {[(consumer_user_defined_category.value) for consumer_user_defined_category in ConsumerUserDefinedCategoryType]}"
-                    )
-
-        return user_defined_category
-
-
-class BaseConsumer(BaseEquipment, ABC):
-    """Base class for all consumers."""
-
-    consumes: ConsumptionType
-    fuel: Optional[dict[Period, FuelType]] = None
-
-    @field_validator("fuel", mode="before")
-    @classmethod
-    def validate_fuel_exist(cls, fuel, info: ValidationInfo):
-        """
-        Make sure fuel is set if consumption type is FUEL.
-        """
-        if isinstance(fuel, dict) and len(fuel.values()) > 0:
-            fuel = _convert_keys_in_dictionary_from_str_to_periods(fuel)
-        if info.data.get("consumes") == ConsumptionType.FUEL and (fuel is None or len(fuel) < 1):
-            msg = f"Missing fuel for fuel consumer '{info.data.get('name')}'"
-            raise ValueError(msg)
-        return fuel
-
-
-class ExpressionTimeSeries(EcalcBaseModel):
-    value: ExpressionType
-    unit: Unit
-    type: Optional[RateType] = None
-
-
-class ExpressionStreamConditions(EcalcBaseModel):
-    rate: Optional[ExpressionTimeSeries] = None
-    pressure: Optional[ExpressionTimeSeries] = None
-    temperature: Optional[ExpressionTimeSeries] = None
-    fluid_density: Optional[ExpressionTimeSeries] = None
+class ExpressionStreamConditions:
+    def __init__(
+        self,
+        rate: Optional[ExpressionTimeSeries] = None,
+        pressure: Optional[ExpressionTimeSeries] = None,
+        temperature: Optional[ExpressionTimeSeries] = None,
+        fluid_density: Optional[ExpressionTimeSeries] = None,
+    ):
+        self.rate = rate
+        self.pressure = pressure
+        self.temperature = temperature
+        self.fluid_density = fluid_density
 
 
 ConsumerID = str
@@ -110,13 +35,13 @@ class ExpressionStreamConditions(EcalcBaseModel):
 SystemStreamConditions = dict[ConsumerID, dict[StreamID, ExpressionStreamConditions]]
 
 
-class Crossover(EcalcBaseModel):
-    model_config = ConfigDict(populate_by_name=True)
-
-    stream_name: Optional[str] = Field(None)
-    from_component_id: str
-    to_component_id: str
+class Crossover:
+    def __init__(self, from_component_id: str, to_component_id: str, stream_name: Optional[str] = None):
+        self.stream_name = stream_name
+        self.from_component_id = from_component_id
+        self.to_component_id = to_component_id
 
 
-class SystemComponentConditions(EcalcBaseModel):
-    crossover: list[Crossover]
+class SystemComponentConditions:
+    def __init__(self, crossover: list[Crossover]):
+        self.crossover = crossover

src/libecalc/domain/infrastructure/energy_components/common.py

@@ -1,10 +1,10 @@
-from typing import Annotated, Literal, Optional, TypeVar, Union
-
-from pydantic import ConfigDict, Field
+from typing import Literal, Optional, TypeVar, Union
 
 from libecalc.common.component_type import ComponentType
+from libecalc.common.consumption_type import ConsumptionType
+from libecalc.common.string.string_utils import generate_id
+from libecalc.common.time_utils import Period
 from libecalc.domain.infrastructure.energy_components.asset.asset import Asset
-from libecalc.domain.infrastructure.energy_components.base.component_dto import BaseConsumer
 from libecalc.domain.infrastructure.energy_components.compressor.component_dto import CompressorComponent
 from libecalc.domain.infrastructure.energy_components.consumer_system.consumer_system_dto import ConsumerSystem
 from libecalc.domain.infrastructure.energy_components.electricity_consumer.electricity_consumer import (
@@ -14,10 +14,11 @@
 from libecalc.domain.infrastructure.energy_components.generator_set.generator_set_dto import GeneratorSet
 from libecalc.domain.infrastructure.energy_components.installation.installation import Installation
 from libecalc.domain.infrastructure.energy_components.pump.component_dto import PumpComponent
-from libecalc.dto.base import EcalcBaseModel
+from libecalc.domain.infrastructure.energy_components.utils import _convert_keys_in_dictionary_from_str_to_periods
+from libecalc.dto.utils.validators import validate_temporal_model
 from libecalc.expression import Expression
 
-Consumer = Annotated[Union[FuelConsumer, ElectricityConsumer], Field(discriminator="consumes")]
+Consumer = Union[FuelConsumer, ElectricityConsumer]
 
 ComponentDTO = Union[
     Asset,
@@ -31,36 +32,79 @@
 ]
 
 
-class CompressorOperationalSettings(EcalcBaseModel):
-    rate: Expression
-    inlet_pressure: Expression
-    outlet_pressure: Expression
-
+class CompressorOperationalSettings:
+    def __init__(self, rate: Expression, inlet_pressure: Expression, outlet_pressure: Expression):
+        self.rate = rate
+        self.inlet_pressure = inlet_pressure
+        self.outlet_pressure = outlet_pressure
 
-class PumpOperationalSettings(EcalcBaseModel):
-    rate: Expression
-    inlet_pressure: Expression
-    outlet_pressure: Expression
-    fluid_density: Expression
 
+class PumpOperationalSettings:
+    def __init__(
+        self, rate: Expression, inlet_pressure: Expression, outlet_pressure: Expression, fluid_density: Expression
+    ):
+        self.rate = rate
+        self.inlet_pressure = inlet_pressure
+        self.outlet_pressure = outlet_pressure
+        self.fluid_density = fluid_density
 
-class Stream(EcalcBaseModel):
-    model_config = ConfigDict(populate_by_name=True)
 
-    stream_name: Optional[str] = Field(None)
-    from_component_id: str
-    to_component_id: str
+class Stream:
+    def __init__(self, from_component_id: str, to_component_id: str, stream_name: Optional[str] = None):
+        self.stream_name = stream_name
+        self.from_component_id = from_component_id
+        self.to_component_id = to_component_id
 
 
 ConsumerComponent = TypeVar("ConsumerComponent", bound=Union[CompressorComponent, PumpComponent])
 
 
-class TrainComponent(BaseConsumer):
-    component_type: Literal[ComponentType.TRAIN_V2] = Field(
-        ComponentType.TRAIN_V2,
-        title="TYPE",
-        description="The type of the component",
-        alias="TYPE",
-    )
-    stages: list[ConsumerComponent]
-    streams: list[Stream]
+class TrainComponent:
+    component_type: Literal[ComponentType.TRAIN_V2] = ComponentType.TRAIN_V2
+
+    def __init__(
+        self,
+        name: str,
+        regularity: dict,
+        consumes,
+        user_defined_category: dict,
+        component_type: ComponentType,
+        stages: list,
+        streams: list,
+    ):
+        self.name = name
+        self.regularity = self.check_regularity(regularity)
+        validate_temporal_model(self.regularity)
+        self.consumes = consumes
+        self.user_defined_category = user_defined_category
+        self.component_type = component_type
+        self.stages = stages
+        self.streams = streams
+
+    @property
+    def id(self) -> str:
+        return generate_id(self.name)
+
+    @staticmethod
+    def check_regularity(regularity: dict[Period, Expression]):
+        if isinstance(regularity, dict) and len(regularity.values()) > 0:
+            regularity = _convert_keys_in_dictionary_from_str_to_periods(regularity)
+        return regularity
+
+    def is_fuel_consumer(self) -> bool:
+        return self.consumes == ConsumptionType.FUEL
+
+    def is_electricity_consumer(self) -> bool:
+        return self.consumes == ConsumptionType.ELECTRICITY
+
+    def is_provider(self) -> bool:
+        return False
+
+    def is_container(self) -> bool:
+        return False
+
+    def get_component_process_type(self) -> ComponentType:
+        return self.component_type
+
+    def get_name(self) -> str:
+        return self.name