process_error.py

from dataclasses import dataclass

from libecalc.common.errors.exceptions import EcalcError
from libecalc.process.process_pipeline.process_unit import ProcessUnitId


@dataclass(frozen=True)
class CompressorOperatingPoint:
    """Thermodynamic and mechanical state describing what a compressor was asked to do.

    Together these fields are sufficient to reproduce the operating point in isolation
    (build the inlet stream, set the speed, request the head/efficiency).
    """

    inlet_pressure_bara: float
    inlet_temperature_kelvin: float
    actual_rate_m3_per_hour: float
    polytropic_head_joule_per_kg: float
    polytropic_efficiency: float
    speed: float


class ProcessError(EcalcError):
    def __init__(self, reason: str | None = None):
        self.reason = reason
        super().__init__(title="Unable to produce an outlet stream", message=reason or "")


class OutsideCapacityError(ProcessError):
    def __init__(self, reason: str = "Operational point is outside capacity."):
        super().__init__(reason)


class RateTooLowError(OutsideCapacityError):
    def __init__(
        self,
        process_unit_id: ProcessUnitId,
        actual_rate: float | None = None,
        boundary_rate: float | None = None,
        reason: str = "Rate is too low.",
    ):
        self.actual_rate = actual_rate
        self.boundary_rate = boundary_rate
        self.process_unit_id = process_unit_id
        super().__init__(reason)


class OutletFluidNotAchievableError(ProcessError):
    """Raised when the compressor cannot produce a thermodynamically valid outlet stream.

    This typically means the EOS / PH flash rejected the requested outlet state
    (e.g. NaN/inf properties, enthalpy did not converge, or NeqSim raised an
    exception). Unlike RateTooHighError the compressor was operating inside its
    chart capacity — the fluid itself is the problem.

    All fields describe the exact operating point at failure, enough to reproduce
    the issue in isolation.
    """

    def __init__(
        self,
        process_unit_id: ProcessUnitId,
        unachievable_operating_point: CompressorOperatingPoint,
        reason: str = "Outlet fluid state is not achievable.",
    ):
        self.process_unit_id = process_unit_id
        self.unachievable_operating_point = unachievable_operating_point
        super().__init__(reason)


class RateTooHighError(OutsideCapacityError):
    def __init__(
        self,
        process_unit_id: ProcessUnitId,
        actual_rate: float | None = None,
        boundary_rate: float | None = None,
        reason: str = "Rate is too high.",
    ):
        self.actual_rate = actual_rate
        self.boundary_rate = boundary_rate
        self.process_unit_id = process_unit_id
        super().__init__(reason)