aspen_pysys.model.unit_operation

16class HysysAdjust(HysysUnitOperation):
17    """Class that represents a Adjust model from the HYSYS app."""
18
19    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
20        super().__init__(connection, obj, HysysUnitOpType.ADJUST)
21
22    def get_target_value(self) -> float:
23        """Get the target value.
24
25        Returns:
26            float: Target value
27        """
28        return self.get_float("TargetValue")
29
30    def set_target_value(self, value: float) -> None:
31        """Set the target value.
32
33        Args:
34            value (float): New value
35        """
36        self.set_float("TargetValue", value)
37
38    def get_tolerance(self) -> float:
39        """Get the tolerance value.
40
41        Returns:
42            float: Tolerance value
43        """
44        return self.get_float("Tolerance")
45
46    def set_tolerance(self, value: float) -> None:
47        """Set the tolerance value.
48
49        Args:
50            value (float): New value
51        """
52        self.set_float("Tolerance", value)
53
54    def get_step_size(self) -> float:
55        """Get the step size.
56
57        Returns:
58            float: Step size
59        """
60        return self.get_float("StepSize")
61
62    def set_step_size(self, value: float) -> None:
63        """Set the step size.
64
65        Args:
66            value (float): New value
67        """
68        self.set_float("StepSize", value)
69
70    def get_max_iterations(self) -> int:
71        """Get the maximum number of iterations.
72
73        Returns:
74            float: Maximum number of iterations
75        """
76        return int(self.get_float("MaximumIterations"))
77
78    def set_max_iterations(self, value: int) -> None:
79        """Set the maximum number of iterations.
80
81        Args:
82            value (float): New value
83        """
84        self.set_float("MaximumIterations", float(value))

19class HysysComponentSplitter(HysysUnitOperation):
20    """Class that represents a Component Splitter model from the HYSYS app."""
21
22    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
23        super().__init__(connection, obj, HysysUnitOpType.COMPONENT_SPLITTER)
24
25    def get_inlets(self) -> HysysDictionary[HysysMaterialStream]:
26        """Get the inlets.
27
28        Returns:
29            HysysDictionary[HysysMaterialStream]: Inlets
30        """
31        return self.get_dict("Inlets").map(HysysModel.FACTORY.get_material_stream)
32
33    def get_overhead_outlets(self) -> HysysDictionary[HysysMaterialStream]:
34        """Get the overhead outlets.
35
36        Returns:
37            HysysDictionary[HysysMaterialStream]: Overhead outlets
38        """
39        return self.get_dict("OverheadOutlets").map(
40            HysysModel.FACTORY.get_material_stream
41        )
42
43    def get_bottoms_outlet(self) -> HysysMaterialStream:
44        """Get the bottoms outlet.
45
46        Returns:
47            HysysMaterialStream: Bottoms outlet
48        """
49        return HysysModel.FACTORY.get_material_stream(
50            self.get_named_obj("BottomOutlets")
51        )
52
53    def get_energy_inlets(self) -> HysysDictionary[HysysEnergyStream]:
54        """Get the energy inlets.
55
56        Returns:
57            HysysDictionary[HysysEnergyStream]: Energy inlets
58        """
59        return self.get_dict("EnergyInlets").map(HysysModel.FACTORY.get_energy_stream)
60
61    def get_split_basis(self) -> HysysArray:
62        """Get the split basis.
63
64        Returns:
65            HysysArray: Split basis
66        """
67        return self.get_array("SplitBasis")
68
69    def get_split_type(self) -> HysysArray:
70        """Get the split type.
71
72        Returns:
73            HysysArray: Split type
74        """
75        return self.get_array("SplitType")

21class HysysCompressor(
22    FeedStreamGetterMixin,
23    ProductStreamGetterMixin,
24    EnergyStreamGetterMixin,
25    HysysUnitOperation,
26):
27    """Class that represents a Compressor model from the HYSYS app."""
28
29    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
30        super().__init__(connection, obj, HysysUnitOpType.COMPRESSOR)
31
32    def get_delta_pressure(self) -> float:
33        """Get the pressure difference.
34
35        Returns:
36            float: Pressure difference
37        """
38        return self.get_float("DeltaP")
39
40    def get_delta_temperature(self) -> float:
41        """Get the temperature difference.
42
43        Returns:
44            float: Temperature difference
45        """
46        return self.get_float("DeltaT")
47
48    def get_duty(self) -> float:
49        """Get the duty.
50
51        Returns:
52            float: Duty
53        """
54        return self.get_energy_stream().heat_flow()

21class HysysControlValve(
22    FeedStreamGetterMixin,
23    ProductStreamGetterMixin,
24    HysysUnitOperation,
25):
26    """Class that represents a Control Valve model from the HYSYS app."""
27
28    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
29        super().__init__(connection, obj, HysysUnitOpType.CONTROL_VALVE)
30
31    def get_pressure_drop(self) -> float:
32        """Get the pressure drop.
33
34        Returns:
35            float: Pressure drop.
36        """
37        return self.get_float("PressureDrop")
38
39    def has_backflow(self) -> bool:
40        """Check if the valve has backflow.
41
42        Returns:
43            bool: If the valve has backflow
44        """
45        return is_negative(self.get_pressure_drop())
46
47    def has_no_pressure_drop(self) -> bool:
48        """Check if the valve has zero pressure drop.
49
50        Returns:
51            bool: If the valve has zero pressure drop
52        """
53        return is_zero(self.get_pressure_drop())

16class HysysConversionReactor(HysysUnitOperation):
17    """Class that represents a Conversion Reactor model from the HYSYS app."""
18
19    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
20        super().__init__(connection, obj, HysysUnitOpType.CONVERSION_REACTOR)

17class HysysCooler(HysysUnitOperation):
18    """Class that represents a Cooler model from the HYSYS app."""
19
20    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
21        super().__init__(connection, obj, HysysUnitOpType.COOLER)
22
23    def get_delta_temperature(self) -> float:
24        """Get the temperature difference.
25
26        Returns:
27            float: Temperature difference
28        """
29        return self.get_float("DeltaT")
30
31    def has_temperature_decrease(self) -> float:
32        """Check if the temperature decreases across the cooler.
33
34        Returns:
35            bool: If the temperature decreases
36        """
37        return is_negative(self.get_delta_temperature())

 26class HysysDistillationColumn(HysysUnitOperation):
 27    """Class that represents a Distillation Column model from the HYSYS app."""
 28
 29    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
 30        super().__init__(connection, obj, HysysUnitOpType.DISTILLATION_COLUMN)
 31
 32    def _get_flowsheet(self) -> HysysObject:
 33        return self.get_obj("ColumnFlowsheet")
 34
 35    def _get_operations(self) -> HysysDictionary:
 36        return self._get_flowsheet().get_dict("Operations")
 37
 38    def _get_main_tower(self) -> HysysNamedObject:
 39        return HysysNamedObject(
 40            self.get_connection(),
 41            self._get_operations().get("Main Tower"),
 42        )
 43
 44    def _get_condenser(self) -> HysysObject:
 45        return self._get_operations().get("Condenser")
 46
 47    def _get_reboiler(self) -> HysysObject:
 48        return self._get_operations().get("Reboiler")
 49
 50    def get_condenser_pressure(self) -> float:
 51        """Get the condenser pressure.
 52
 53        Returns:
 54            float: Condenser pressure
 55        """
 56        return self._get_condenser().get_float("VesselPressure")
 57
 58    def get_condenser_duty(self) -> float:
 59        """Get the condenser duty.
 60
 61        Returns:
 62            float: Condenser duty
 63        """
 64        return self._get_condenser().get_float("HeatFlow")
 65
 66    def get_reboiler_pressure(self) -> float:
 67        """Get the reboiler pressure.
 68
 69        Returns:
 70            float: Reboiler pressure
 71        """
 72        return self._get_reboiler().get_float("VesselPressure")
 73
 74    def get_reboiler_duty(self) -> float:
 75        """Get the reboiler duty.
 76
 77        Returns:
 78            float: Reboiler duty
 79        """
 80        return self._get_reboiler().get_float("HeatFlow")
 81
 82    def get_pressure_diff(self) -> float:
 83        """Get the pressure difference between the reboiler and the condenser.
 84
 85        Returns:
 86            float: Pressure difference
 87        """
 88        return self.get_reboiler_pressure() - self.get_condenser_pressure()
 89
 90    def is_converged(self) -> bool:
 91        """Check if the column has converged.
 92
 93        Returns:
 94            bool: If the column has converged.
 95        """
 96        return self._get_flowsheet().get_bool("CfsConverged")
 97
 98    def run(self, max_iterations: int = 10000) -> None:
 99        """Run the column.
100
101        Args:
102            max_iterations (int, optional): Maximum number of iterations. Defaults to 10000.
103        """  # noqa: E501
104        self._get_flowsheet().set_int("MaximumIterations", max_iterations)
105
106        run_func = self._get_flowsheet().get_func("Run")
107        run_func.call()
108
109    def reset(self) -> None:
110        """Reset the column flowsheet."""
111        func = self._get_flowsheet().get_func("Reset")
112        func.call()
113
114    def get_attached_material_feeds(self) -> dict[str, HysysMaterialStream]:
115        """Get a dictionary containing the names of the attached material feeds as keys and the streams as values.
116
117        Returns:
118            dict[str, HysysMaterialStream]: Attached material feeds.
119        """  # ruff:ignore[line-too-long]
120        return {
121            key: HysysModel.FACTORY.get_material_stream(value)
122            for key, value in (
123                self.get_attached_feeds()
124                .map(HysysModel.FACTORY.get_process_stream)
125                .filter(
126                    lambda _, stream: (
127                        stream.get_stream_type() == HysysStreamType.MATERIAL_STREAM
128                    ),
129                )
130                .items()
131            )
132        }
133
134    def get_subflowsheet_material_feeds(self) -> dict[str, HysysMaterialStream]:
135        """Get a dictionary containing the names of the attached material feeds in the column subflowsheet as keys and the streams as values.
136
137        Returns:
138            dict[str, HysysMaterialStream]: Attached material feeds in the column subflowsheet
139        """  # ruff:ignore[line-too-long]
140        return {
141            key: HysysModel.FACTORY.get_material_stream(value)
142            for key, value in (
143                self._get_flowsheet()
144                .get_dict("FeedStreams")
145                .map(HysysModel.FACTORY.get_process_stream)
146                .filter(
147                    lambda _, stream: (
148                        stream.get_stream_type() == HysysStreamType.MATERIAL_STREAM
149                    ),
150                )
151                .items()
152            )
153        }
154
155    def specify_feed_location(self, feed: str | HysysObject, tray: int) -> None:
156        """Specify the tray number for a given feed with respect to the column subflowsheet.
157
158        Args:
159            feed (str | HysysObject): Either the feed name or feed object
160            tray (int): Tray number
161
162        Raises:
163            PysysError: Feed does not exist in the column subflowsheet
164        """  # ruff:ignore[line-too-long]
165        if isinstance(feed, str):
166            feed_obj = self.get_subflowsheet_material_feeds().get(feed)
167
168            if feed_obj is None:
169                message = (
170                    f"Feed stream named '{feed}' does not exist in the column subflowsheet"  # ruff:ignore[line-too-long]
171                    f" (available feeds: {self.get_subflowsheet_material_feeds().keys()})."  # ruff:ignore[line-too-long]
172                )
173                raise PysysError(message)
174
175            feed = feed_obj
176
177        with feed as feed_com_obj:
178            func = self._get_main_tower().get_func("SpecifyFeedLocation")
179            func.call(feed_com_obj, tray)
180
181    def add_material_feed(self, feed_name: str, tray: int) -> None:
182        """Add a feed stream.
183
184        Args:
185            feed_name (str): Stream name
186            tray (int): Tray number
187        """
188        self._get_main_tower().get_func("AddFeedStream").call(feed_name, tray, False)  # noqa: FBT003
189
190    def _confirm_main_and_subflowsheet_feeds_match(self) -> None:
191        flowsheet_feed_names: set[str] = set(self.get_attached_material_feeds().keys())
192        subflowsheet_feed_names: set[str] = set(
193            self.get_subflowsheet_material_feeds().keys()
194        )
195
196        if flowsheet_feed_names != subflowsheet_feed_names:
197            flowsheet_feed_names_str = ", ".join(flowsheet_feed_names)
198            subflowsheet_feed_names_str = ", ".join(subflowsheet_feed_names)
199
200            error_message = (
201                "Feed trays cannot be identified, possibly due to "
202                "a mismatch in stream names between the main and column flowsheets."
203                f"\nMain = {flowsheet_feed_names_str}"
204                f", Column = {subflowsheet_feed_names_str}."
205            )
206
207            raise PysysError(error_message)
208
209    def get_feed_tray_numbers(self) -> dict[str, int]:
210        """Get the feed tray numbers.
211
212        Returns:
213            dict[str, int]: Dictionary containing the feed names as keys and their respective tray numbers as values.
214        """  # ruff:ignore[line-too-long]
215        self._confirm_main_and_subflowsheet_feeds_match()
216
217        feed_trays_hysys_dict = self._get_main_tower().get_dict("FeedStages")
218        feed_trays_dict: dict[str, int] = {}
219
220        for key in feed_trays_hysys_dict:
221            feed_tray = feed_trays_hysys_dict.get(key)
222            streams_at_feed_tray = feed_tray.get_dict("FeedStreams")
223
224            if len(streams_at_feed_tray) == 0:
225                continue
226
227            feed_tray_number = feed_tray.get_int("IFaceStageNumber")
228
229            for stream_name in streams_at_feed_tray:
230                feed_trays_dict.update({stream_name: feed_tray_number})
231
232        return feed_trays_dict
233
234    def get_tray_count(self) -> int:
235        """Get the tray count.
236
237        Returns:
238            int: Tray count
239        """
240        return int(self._get_main_tower().get_float("NumberOfTrays"))
241
242    def set_tray_count(self, count: int) -> None:
243        """Set the tray count.
244
245        Args:
246            count (int): New value
247
248        Raises:
249            PysysError: Tray count is lesser than 3.
250        """
251        if count < MIN_TRAY_COUNT:
252            message = f"Tray count must not be lesser than {MIN_TRAY_COUNT}."
253            raise PysysError(message)
254
255        self._get_main_tower().set_float("NumberOfTrays", count)
256
257    def get_current_iteration(self) -> int:
258        """Get the number of finished iterations.
259
260        Returns:
261            int: Current iteration
262        """
263        return self._get_flowsheet().get_int("CurrentIteration") - 1
264
265    def _get_column_trays(self) -> HysysDictionary:
266        return self._get_flowsheet().get_dict("ColumnStages")
267
268    def _get_separation_tray(self, position: int) -> HysysObject:
269        main_tower_name = self._get_main_tower().get_name()
270        tray_name = f"{position}__{main_tower_name}"
271
272        return self._get_column_trays().get(tray_name).get_obj("SeparationStage")

16class HysysFiredHeater(HysysUnitOperation):
17    """Class that represents a Fired Heater model from the HYSYS app."""
18
19    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
20        super().__init__(connection, obj, HysysUnitOpType.FIRED_HEATER)

 23class HysysHeatExchanger(HysysUnitOperation):
 24    """Class that represents a Heat Exchanger model from the HYSYS app."""
 25
 26    def __init__(self, connection: HysysCase, obj: HysysNamedObjReadable) -> None:
 27        super().__init__(connection, obj, HysysUnitOpType.HEAT_EXCHANGER)
 28
 29    @override
 30    def get_attached_feeds(self) -> HysysDictionary[HysysMaterialStream]:
 31        return (
 32            super()
 33            .get_attached_feeds()
 34            .map(
 35                HysysModel.FACTORY.get_material_stream,
 36            )
 37        )
 38
 39    @override
 40    def get_attached_products(self) -> HysysDictionary[HysysMaterialStream]:
 41        return (
 42            super()
 43            .get_attached_products()
 44            .map(
 45                HysysModel.FACTORY.get_material_stream,
 46            )
 47        )
 48
 49    def get_duty(self) -> float:
 50        """Get the duty.
 51
 52        Returns:
 53            float: Duty
 54        """
 55        return self.get_float("Duty")
 56
 57    def get_lmtd(self) -> float:
 58        """Get the log mean temperature difference (LMTD).
 59
 60        Returns:
 61            float: LMTD
 62        """
 63        return self.get_float("LMTD")
 64
 65    def get_tube_side_feed(self) -> HysysMaterialStream:
 66        """Get the tube side's feed stream.
 67
 68        Returns:
 69            HysysMaterialStream: Tube side's feed stream
 70        """
 71        return HysysModel.FACTORY.get_material_stream(
 72            self.get_named_obj("TubeSideFeed")
 73        )
 74
 75    def get_tube_side_product(self) -> HysysMaterialStream:
 76        """Get the tube side's product stream.
 77
 78        Returns:
 79            HysysMaterialStream: Tube side's product stream
 80        """
 81        return HysysModel.FACTORY.get_material_stream(
 82            self.get_named_obj("TubeSideProduct")
 83        )
 84
 85    def get_shell_side_feed(self) -> HysysMaterialStream:
 86        """Get the shell side's feed stream.
 87
 88        Returns:
 89            HysysMaterialStream: Shell side's feed stream
 90        """
 91        return HysysModel.FACTORY.get_material_stream(
 92            self.get_named_obj("ShellSideFeed")
 93        )
 94
 95    def get_shell_side_product(self) -> HysysMaterialStream:
 96        """Get the shell side's product stream.
 97
 98        Returns:
 99            HysysMaterialStream: Shell side's product stream
100        """
101        return HysysModel.FACTORY.get_material_stream(
102            self.get_named_obj("ShellSideProduct")
103        )
104
105    def is_tube_side_hot(self) -> bool:
106        """Check if the tube side is the hot side.
107
108        Returns:
109            bool: If the tube side is the hot side
110        """
111        tube_side_feed_temp = self.get_tube_side_feed().get_temperature()
112        tube_side_product_temp = self.get_tube_side_product().get_temperature()
113
114        return tube_side_feed_temp > tube_side_product_temp
115
116    def get_tube_heat_transf_coef(self) -> float:
117        """Get the tube side's heat transfer coefficient.
118
119        Returns:
120            float: Tube side's heat transfer coefficient.
121        """
122        return self.get_float("TubeHeatTransferCoef")
123
124    def set_tube_heat_transf_coef(self, value: float) -> None:
125        """Set the tube side's heat transfer coefficient.
126
127        Args:
128            value (float): New value.
129        """
130        return self.set_float("TubeHeatTransferCoef", value)
131
132    def get_shell_heat_transf_coef(self) -> float:
133        """Get the shell side's heat transfer coefficient.
134
135        Returns:
136            float: Shell side's heat transfer coefficient.
137        """
138        return self.get_float("ShellHeatTransferCoef")
139
140    def set_shell_heat_transf_coef(self, value: float) -> None:
141        """Set the shell side's heat transfer coefficient.
142
143        Args:
144            value (float): New value.
145        """
146        return self.set_float("ShellHeatTransferCoef", value)
147
148    def get_sides_dict(self) -> HysysHXSidesDict:
149        """Get a dictionary containing the hot and cold sides' feed and product streams.
150
151        Returns:
152            HysysHXSidesDict: Hot and cold sides' feed and product streams
153        """
154        if self.is_tube_side_hot():
155            return {
156                "Hot": (self.get_tube_side_feed(), self.get_tube_side_product()),
157                "Cold": (self.get_shell_side_feed(), self.get_shell_side_product()),
158            }
159
160        return {
161            "Hot": (self.get_shell_side_feed(), self.get_shell_side_product()),
162            "Cold": (self.get_tube_side_feed(), self.get_tube_side_product()),
163        }