diff --git a/src/tespy/components/__init__.py b/src/tespy/components/__init__.py index c450e1aa..b4e65bfe 100644 --- a/src/tespy/components/__init__.py +++ b/src/tespy/components/__init__.py @@ -36,6 +36,10 @@ from .power.source import PowerSource # noqa: F401 from .reactors.fuel_cell import FuelCell # noqa: F401 from .reactors.water_electrolyzer import WaterElectrolyzer # noqa: F401 +from .sorption.sorption import Absorber # noqa: F401 +from .sorption.sorption import CooledAbsorber # noqa: F401 +from .sorption.sorption import Desorber # noqa: F401 +from .sorption.sorption import HeatedDesorber # noqa: F401 from .subsystem import Subsystem # noqa: F401 from .turbomachinery.compressor import Compressor # noqa: F401 from .turbomachinery.pump import Pump # noqa: F401 diff --git a/src/tespy/components/sorption/__init__.py b/src/tespy/components/sorption/__init__.py new file mode 100644 index 00000000..cf09fc14 --- /dev/null +++ b/src/tespy/components/sorption/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 diff --git a/src/tespy/components/sorption/sorption.py b/src/tespy/components/sorption/sorption.py new file mode 100644 index 00000000..ed85a805 --- /dev/null +++ b/src/tespy/components/sorption/sorption.py @@ -0,0 +1,903 @@ +# -*- coding: utf-8 + +"""Module for Absorber and Desorber sorption components. + + +This file is part of project TESPy (github.com/oemof/tespy). It's copyrighted +by the contributors recorded in the version control history of the file, +available from its original location tespy/components/sorption/sorption.py + +SPDX-License-Identifier: MIT +""" + +from tespy.components.component import component_registry +from tespy.components.nodes.base import NodeBase +from tespy.tools.data_containers import ComponentMandatoryConstraints as dc_cmc +from tespy.tools.data_containers import ComponentProperties as dc_cp +from tespy.tools.fluid_properties.mixtures import _xi_sat_libr + +_LIBR_KEY = "LiBr" + + +class _SorptionBase(NodeBase): + r""" + Shared equations for absorption-cycle sorption components. + + Both :class:`Absorber` and :class:`Desorber` enforce: + + - total mass balance (from :class:`.NodeBase`) + - LiBr mass balance between solution connections + - pressure equality across all ports (from :class:`.NodeBase`) + - saturation condition at the solution outlet + + The optional heat-flow parameter :code:`Q` adds an energy balance + equation when set. + """ + + def get_parameters(self): + return { + "Q": dc_cp( + func=self.heat_func, + deriv=self.heat_deriv, + num_eq=1, + description="heat transferred into the component", + ) + } + + # ------------------------------------------------------------------ + # LiBr mass balance: m_sol_in * xi_in = m_sol_out * xi_out + # ------------------------------------------------------------------ + + def libr_balance_func(self): + sol_in = self._solution_inlet() + sol_out = self._solution_outlet() + return ( + sol_in.m.val_SI * sol_in.fluid.val[_LIBR_KEY] + - sol_out.m.val_SI * sol_out.fluid.val[_LIBR_KEY] + ) + + def libr_balance_dependents(self): + sol_in = self._solution_inlet() + sol_out = self._solution_outlet() + return { + "scalars": [[sol_in.m, sol_out.m]], + "vectors": [{ + sol_in.fluid: {_LIBR_KEY} & sol_in.fluid.is_var, + sol_out.fluid: {_LIBR_KEY} & sol_out.fluid.is_var, + }], + } + + # ------------------------------------------------------------------ + # Saturation condition at solution outlet: xi_out = xi_sat(p_out, T_out) + # ------------------------------------------------------------------ + + def saturation_func(self): + sol_out = self._solution_outlet() + T = sol_out.calc_T() + return _xi_sat_libr(sol_out.p.val_SI, T, sol_out.fluid_data) - sol_out.fluid.val[_LIBR_KEY] + + def saturation_dependents(self): + sol_out = self._solution_outlet() + return { + "scalars": [[sol_out.p, sol_out.h]], + "vectors": [{sol_out.fluid: {_LIBR_KEY} & sol_out.fluid.is_var}], + } + + # ------------------------------------------------------------------ + # Binary fluid balance at solution outlet: xi_LiBr + x_H2O = 1 + # ------------------------------------------------------------------ + + def fluid_balance_func(self): + sol_out = self._solution_outlet() + return sol_out.fluid.val["H2O"] + sol_out.fluid.val[_LIBR_KEY] - 1.0 + + def fluid_balance_dependents(self): + sol_out = self._solution_outlet() + return { + "scalars": [[]], + "vectors": [{sol_out.fluid: {"H2O", _LIBR_KEY} & sol_out.fluid.is_var}], + } + + # ------------------------------------------------------------------ + # Energy balance (optional) + # ------------------------------------------------------------------ + + def heat_func(self): + return self.Q.val_SI - self.calc_Q() + + def heat_deriv(self, increment_filter, k, dependents=None): + for c in self.inl: + if c.m.is_var: + self.jacobian[k, c.m.J_col] = -c.h.val_SI + if c.h.is_var: + self.jacobian[k, c.h.J_col] = -c.m.val_SI + for c in self.outl: + if c.m.is_var: + self.jacobian[k, c.m.J_col] = c.h.val_SI + if c.h.is_var: + self.jacobian[k, c.h.J_col] = c.m.val_SI + + def calc_Q(self): + return ( + sum(c.h.val_SI * c.m.val_SI for c in self.outl) + - sum(c.h.val_SI * c.m.val_SI for c in self.inl) + ) + + def _update_num_eq(self): + sol_out = self._solution_outlet() + num_eq = 0 if len(sol_out.fluid.is_var) == 0 else 1 + self.constraints["fluid_balance_constraints"].num_eq = num_eq + + def calc_parameters(self): + self.Q.val = self.calc_Q() + + def convergence_check(self): + from tespy.tools.fluid_properties.functions import h_mix_pT + sol_out = self._solution_outlet() + if not sol_out.h.is_var: + return + try: + T = sol_out.calc_T() + except Exception: + T = 250 + if T < 274 or T > 499: + try: + h = h_mix_pT( + sol_out.p.val_SI, 320, sol_out.fluid_data, sol_out.mixing_rule + ) + sol_out.h.set_reference_val_SI(h) + except Exception: + pass + + def propagate_to_target(self, branch): + return + + +@component_registry +class Absorber(_SorptionBase): + r""" + Ideal single-effect absorber for LiBr-water absorption cycles. + + The absorber merges a water-vapour stream (:code:`in2`) into a poor + LiBr solution (:code:`in1`) and produces a rich solution + (:code:`out1`). The outlet solution is assumed to be at thermodynamic + equilibrium (saturation condition). + + Ports + ----- + - :code:`in1` - poor LiBr solution (SolutionConnection) + - :code:`in2` - water vapour (Connection) + - :code:`out1` - rich LiBr solution (SolutionConnection) + + Mandatory Equations + ------------------- + - mass balance: :math:`\dot m_{in,1} + \dot m_{in,2} = \dot m_{out,1}` + - LiBr balance: + :math:`\dot m_{in,1} \xi_{in,1} = \dot m_{out,1} \xi_{out,1}` + - pressure equality: all ports at the same pressure + - saturation: + :math:`\xi_{out,1} = \xi_\text{sat}(p_{out,1},\, T_{out,1})` + + Parameters + ---------- + Q : float + Heat removed from the absorber (W); negative by convention. + """ + + @staticmethod + def inlets(): + return ["in1", "in2"] + + @staticmethod + def outlets(): + return ["out1"] + + def _solution_inlet(self): + return self.inl[0] + + def _solution_outlet(self): + return self.outl[0] + + def get_mandatory_constraints(self): + return { + "mass_flow_constraints": dc_cmc( + num_eq_sets=1, + func=self.mass_flow_func, + dependents=self.mass_flow_dependents, + description="mass balance", + ), + "libr_balance_constraints": dc_cmc( + num_eq_sets=1, + func=self.libr_balance_func, + dependents=self.libr_balance_dependents, + description="LiBr mass balance", + ), + "pressure_constraints": dc_cmc( + num_eq_sets=2, + structure_matrix=self.pressure_structure_matrix, + description="pressure equality", + ), + "saturation_constraints": dc_cmc( + num_eq_sets=1, + func=self.saturation_func, + dependents=self.saturation_dependents, + description="saturation at solution outlet", + ), + "fluid_balance_constraints": dc_cmc( + num_eq_sets=1, + func=self.fluid_balance_func, + dependents=self.fluid_balance_dependents, + description="binary fluid balance at solution outlet", + ), + } + + def propagate_wrapper_to_target(self, branch): + inconn = branch["connections"][-1] + if inconn == self.inl[1]: + branch["components"] += [self] + return + if self in branch["components"]: + return + outconn = self.outl[0] + branch["connections"] += [outconn] + branch["components"] += [self] + outconn.target.propagate_wrapper_to_target(branch) + + +@component_registry +class Desorber(_SorptionBase): + + _is_wrapper_branch_source = True + r""" + Ideal single-effect desorber (generator) for LiBr-water absorption cycles. + + The desorber heats a rich LiBr solution (:code:`in1`) and separates it + into a poor solution (:code:`out1`) and water vapour (:code:`out2`). + The poor solution outlet is assumed to be at thermodynamic equilibrium + (saturation condition). + + Ports + ----- + - :code:`in1` - rich LiBr solution (SolutionConnection) + - :code:`out1` - poor LiBr solution (SolutionConnection) + - :code:`out2` - water vapour (Connection) + + Mandatory Equations + ------------------- + - mass balance: :math:`\dot m_{in,1} = \dot m_{out,1} + \dot m_{out,2}` + - LiBr balance: + :math:`\dot m_{in,1} \xi_{in,1} = \dot m_{out,1} \xi_{out,1}` + - pressure equality: all ports at the same pressure + - saturation: + :math:`\xi_{out,1} = \xi_\text{sat}(p_{out,1},\, T_{out,1})` + + Parameters + ---------- + Q : float + Heat supplied to the desorber (W); positive by convention. + """ + + @staticmethod + def inlets(): + return ["in1"] + + @staticmethod + def outlets(): + return ["out1", "out2"] + + def _solution_inlet(self): + return self.inl[0] + + def _solution_outlet(self): + return self.outl[0] + + def get_mandatory_constraints(self): + return { + "mass_flow_constraints": dc_cmc( + num_eq_sets=1, + func=self.mass_flow_func, + dependents=self.mass_flow_dependents, + description="mass balance", + ), + "libr_balance_constraints": dc_cmc( + num_eq_sets=1, + func=self.libr_balance_func, + dependents=self.libr_balance_dependents, + description="LiBr mass balance", + ), + "pressure_constraints": dc_cmc( + num_eq_sets=2, + structure_matrix=self.pressure_structure_matrix, + description="pressure equality", + ), + "saturation_constraints": dc_cmc( + num_eq_sets=1, + func=self.saturation_func, + dependents=self.saturation_dependents, + description="saturation at solution outlet", + ), + "fluid_balance_constraints": dc_cmc( + num_eq_sets=1, + func=self.fluid_balance_func, + dependents=self.fluid_balance_dependents, + description="binary fluid balance at solution outlet", + ), + } + + def convergence_check(self): + super().convergence_check() + vap_out = self.outl[1] + if not vap_out.h.is_var: + return + try: + from tespy.tools.fluid_properties.functions import h_mix_pQ + from tespy.tools.fluid_properties.functions import phase_mix_ph + phase = phase_mix_ph(vap_out.p.val_SI, vap_out.h.val_SI, vap_out.fluid_data) + if phase != "g": + h_g = h_mix_pQ(vap_out.p.val_SI, 1.0, vap_out.fluid_data) + vap_out.h.set_reference_val_SI(h_g) + except Exception: + pass + + def start_fluid_wrapper_branch(self): + vap_conn = self.outl[1] + vap_branch = { + "connections": [vap_conn], + "components": [self], + } + vap_conn.target.propagate_wrapper_to_target(vap_branch) + + sol_conn = self.outl[0] + sol_branch = { + "connections": [sol_conn], + "components": [self], + } + sol_conn.target.propagate_wrapper_to_target(sol_branch) + + return {vap_conn.label: vap_branch, sol_conn.label: sol_branch} + + def propagate_wrapper_to_target(self, branch): + if self in branch["components"]: + return + outconn = self.outl[0] + branch["connections"] += [outconn] + branch["components"] += [self] + outconn.target.propagate_wrapper_to_target(branch) + + +class _TwoSidedSorptionBase(_SorptionBase): + r""" + Shared equations for sorption components with an integrated heat + transfer fluid side. + + The sorption side equations of :class:`_SorptionBase` are restricted to + the sorption ports. On top of those, the heat transfer fluid side adds: + + - mass flow and fluid composition equality between the heat transfer + fluid inlet and outlet + - an overall energy balance (the component itself is adiabatic, all heat + is exchanged between the two sides) + + By convention the heat transfer fluid enters at :code:`in2` and leaves + at :code:`out2` on both components. Optional parameters are the heat + flow :code:`Q` (transferred from the hot side, thus always negative in + analogy to the two-sided heat exchangers), the terminal temperature + differences :code:`ttd_u` and :code:`ttd_l` and the heat transfer fluid + side pressure specifications :code:`pr2` and :code:`dp2`. + """ + + def get_parameters(self): + return { + "Q": dc_cp( + max_val=0, + func=self.heat_func, + dependents=self.heat_dependents, + num_eq_sets=1, + quantity="heat", + description="heat transferred from the hot side", + ), + "ttd_u": dc_cp( + min_val=0, + func=self.ttd_u_func, + dependents=self.ttd_u_dependents, + num_eq_sets=1, + quantity="temperature_difference", + description="upper terminal temperature difference", + ), + "ttd_l": dc_cp( + min_val=0, + func=self.ttd_l_func, + dependents=self.ttd_l_dependents, + num_eq_sets=1, + quantity="temperature_difference", + description="lower terminal temperature difference", + ), + "pr2": dc_cp( + min_val=1e-4, max_val=1, num_eq_sets=1, + structure_matrix=self.pr_structure_matrix, + func_params={"pr": "pr2", "inconn": 1, "outconn": 1}, + quantity="ratio", + description="heat transfer fluid side outlet to inlet pressure ratio", + ), + "dp2": dc_cp( + min_val=0, max_val=1e15, num_eq_sets=1, + structure_matrix=self.dp_structure_matrix, + func_params={"dp": "dp2", "inconn": 1, "outconn": 1}, + quantity="pressure_difference", + description="heat transfer fluid side inlet to outlet absolute pressure change", + ), + } + + def get_mandatory_constraints(self): + return { + "mass_flow_constraints": dc_cmc( + num_eq_sets=1, + func=self.mass_flow_func, + dependents=self.mass_flow_dependents, + description="sorption side mass balance", + ), + "htf_mass_flow_constraints": dc_cmc( + num_eq_sets=1, + structure_matrix=self.htf_variable_equality_structure_matrix, + func_params={"variable": "m"}, + description="heat transfer fluid side mass flow equality", + ), + "htf_fluid_constraints": dc_cmc( + num_eq_sets=1, + structure_matrix=self.htf_variable_equality_structure_matrix, + func_params={"variable": "fluid"}, + description="heat transfer fluid side fluid composition equality", + ), + "libr_balance_constraints": dc_cmc( + num_eq_sets=1, + func=self.libr_balance_func, + dependents=self.libr_balance_dependents, + description="LiBr mass balance", + ), + "pressure_constraints": dc_cmc( + num_eq_sets=2, + structure_matrix=self.pressure_structure_matrix, + description="sorption side pressure equality", + ), + "saturation_constraints": dc_cmc( + num_eq_sets=1, + func=self.saturation_func, + dependents=self.saturation_dependents, + description="saturation at solution outlet", + ), + "fluid_balance_constraints": dc_cmc( + num_eq_sets=1, + func=self.fluid_balance_func, + dependents=self.fluid_balance_dependents, + description="binary fluid balance at solution outlet", + ), + "energy_balance_constraints": dc_cmc( + num_eq_sets=1, + func=self.energy_balance_func, + dependents=self.energy_balance_dependents, + description="energy balance between both sides", + ), + } + + def mass_flow_func(self): + res = 0 + for c in self._sorption_inlets(): + res += c.m.val_SI + for c in self._sorption_outlets(): + res -= c.m.val_SI + return res + + def mass_flow_dependents(self): + return [c.m for c in self._sorption_inlets() + self._sorption_outlets()] + + def htf_variable_equality_structure_matrix(self, k, **kwargs): + variable = kwargs.get("variable") + self._structure_matrix[k, self.inl[1].get_attr(variable).sm_col] = 1 + self._structure_matrix[k, self.outl[1].get_attr(variable).sm_col] = -1 + + def pressure_structure_matrix(self, k): + conns = self._sorption_inlets() + self._sorption_outlets() + first = conns[0] + for eq, conn in enumerate(conns[1:]): + self._structure_matrix[k + eq, first.p.sm_col] = 1 + self._structure_matrix[k + eq, conn.p.sm_col] = -1 + + def energy_balance_func(self): + return ( + sum(c.h.val_SI * c.m.val_SI for c in self.outl) + - sum(c.h.val_SI * c.m.val_SI for c in self.inl) + ) + + def energy_balance_dependents(self): + return [v for c in self.inl + self.outl for v in (c.m, c.h)] + + def heat_dependents(self): + return [self.inl[1].m, self.inl[1].h, self.outl[1].h] + + def _separate_flat_enthalpy_starts(self, seeded): + i = self.inl[1] + o = self.outl[1] + if i.h._reference_container is o.h._reference_container: + return 0 + delta = 1e3 + if abs(o.h.val_SI - i.h.val_SI) >= delta: + return 0 + if o.h.is_var and o.h._reference_container not in seeded: + o.h.set_reference_val_SI(i.h.val_SI + self._htf_dh_sign * delta) + elif i.h.is_var and i.h._reference_container not in seeded: + i.h.set_reference_val_SI(o.h.val_SI - self._htf_dh_sign * delta) + else: + return 0 + return 1 + + def _initial_affine_edges(self): + sorption_ports = self._sorption_inlets() + self._sorption_outlets() + first = sorption_ports[0] + edges = [] + for c in sorption_ports[1:]: + edges += [ + (first.p, c.p, 1.0, 0.0), + (first.h, c.h, 1.0, 0.0), + ] + edges += [ + (self.inl[1].p, self.outl[1].p, 1.0, 0.0), + (self.inl[1].h, self.outl[1].h, 1.0, 0.0), + ] + return edges + + def calc_parameters(self): + self.Q.val_SI = self.calc_Q() + self.pr2.val_SI = self.outl[1].p.val_SI / self.inl[1].p.val_SI + self.dp2.val_SI = self.inl[1].p.val_SI - self.outl[1].p.val_SI + + +@component_registry +class CooledAbsorber(_TwoSidedSorptionBase): + r""" + Absorber with integrated cooling fluid side for LiBr-water absorption + cycles. + + Like :class:`Absorber`, the component merges a water-vapour stream + (:code:`in3`) into a poor LiBr solution (:code:`in1`) and produces a + saturated rich solution (:code:`out1`). The heat of absorption is + transferred to a cooling fluid (:code:`in2` to :code:`out2`) via a + mandatory energy balance, so no external heat flow specification is + required. + + Ports + ----- + - :code:`in1` - poor LiBr solution (SolutionConnection) + - :code:`in2` - cooling fluid (Connection) + - :code:`in3` - water vapour (Connection) + - :code:`out1` - rich LiBr solution (SolutionConnection) + - :code:`out2` - cooling fluid (Connection) + + .. note:: + + The port numbering deviates from :class:`Absorber`: the cooling + fluid occupies :code:`in2` and :code:`out2` in analogy to the cold + side of the two-sided heat exchangers (matching :code:`pr2` and + :code:`dp2`), therefore the water vapour inlet moves to + :code:`in3`. + + Mandatory Equations + ------------------- + - sorption side mass balance: + :math:`\dot m_{in,1} + \dot m_{in,3} = \dot m_{out,1}` + - cooling fluid mass flow and fluid composition equality + - LiBr balance: + :math:`\dot m_{in,1} \xi_{in,1} = \dot m_{out,1} \xi_{out,1}` + - pressure equality of the sorption side ports + - saturation: + :math:`\xi_{out,1} = \xi_\text{sat}(p_{out,1},\, T_{out,1})` + - energy balance: + :math:`0 = \sum \dot m_{out} h_{out} - \sum \dot m_{in} h_{in}` + + Parameters + ---------- + Q : float + Heat transferred from the sorption (hot) side to the cooling fluid + (W); always negative. + + ttd_u : float + Upper terminal temperature difference + :math:`T_{in,1} - T_{out,2}` (K). + + ttd_l : float + Lower terminal temperature difference + :math:`T_{out,1} - T_{in,2}` (K). + + pr2 : float + Cooling fluid side outlet to inlet pressure ratio. + + dp2 : float + Cooling fluid side inlet to outlet absolute pressure change. + """ + + _htf_dh_sign = 1 + + @staticmethod + def inlets(): + return ["in1", "in2", "in3"] + + @staticmethod + def outlets(): + return ["out1", "out2"] + + def _sorption_inlets(self): + return [self.inl[0], self.inl[2]] + + def _sorption_outlets(self): + return [self.outl[0]] + + def _solution_inlet(self): + return self.inl[0] + + def _solution_outlet(self): + return self.outl[0] + + def heat_func(self): + i = self.inl[1] + o = self.outl[1] + return self.Q.val_SI + i.m.val_SI * (o.h.val_SI - i.h.val_SI) + + def calc_Q(self): + i = self.inl[1] + o = self.outl[1] + return -i.m.val_SI * (o.h.val_SI - i.h.val_SI) + + def ttd_u_func(self): + T_sol_in = self.inl[0].calc_T() + T_cool_out = self.outl[1].calc_T() + return self.ttd_u.val_SI - T_sol_in + T_cool_out + + def ttd_u_dependents(self): + sol_in = self.inl[0] + return { + "scalars": [[ + sol_in.p, sol_in.h, self.outl[1].p, self.outl[1].h + ]], + "vectors": [{sol_in.fluid: {_LIBR_KEY} & sol_in.fluid.is_var}], + } + + def ttd_l_func(self): + T_sol_out = self.outl[0].calc_T() + T_cool_in = self.inl[1].calc_T() + return self.ttd_l.val_SI - T_sol_out + T_cool_in + + def ttd_l_dependents(self): + sol_out = self.outl[0] + return { + "scalars": [[ + sol_out.p, sol_out.h, self.inl[1].p, self.inl[1].h + ]], + "vectors": [{sol_out.fluid: {_LIBR_KEY} & sol_out.fluid.is_var}], + } + + def _initial_temperature_edges(self): + sorption_ports = self._sorption_inlets() + self._sorption_outlets() + first = sorption_ports[0] + edges = [(first, c, 0.0, 1.0) for c in sorption_ports[1:]] + edges += [(self.inl[1], self.outl[1], 0.0, 1.0)] + ttd_upper = self.ttd_u.val_SI if self.ttd_u.is_set else 10.0 + weight_upper = 5.0 if self.ttd_u.is_set else 0.3 + ttd_lower = self.ttd_l.val_SI if self.ttd_l.is_set else 10.0 + weight_lower = 5.0 if self.ttd_l.is_set else 0.3 + edges += [ + (self.outl[1], self.inl[0], ttd_upper, weight_upper), + (self.inl[1], self.outl[0], ttd_lower, weight_lower), + ] + return edges + + def calc_parameters(self): + super().calc_parameters() + self.ttd_u.val_SI = self.inl[0].T.val_SI - self.outl[1].T.val_SI + self.ttd_l.val_SI = self.outl[0].T.val_SI - self.inl[1].T.val_SI + + def propagate_wrapper_to_target(self, branch): + inconn = branch["connections"][-1] + if inconn == self.inl[1]: + outconn = self.outl[1] + branch["connections"] += [outconn] + branch["components"] += [self] + outconn.target.propagate_wrapper_to_target(branch) + return + if inconn == self.inl[2]: + branch["components"] += [self] + return + if self in branch["components"]: + return + outconn = self.outl[0] + branch["connections"] += [outconn] + branch["components"] += [self] + outconn.target.propagate_wrapper_to_target(branch) + + +@component_registry +class HeatedDesorber(_TwoSidedSorptionBase): + r""" + Desorber (generator) with integrated heating fluid side for LiBr-water + absorption cycles. + + Like :class:`Desorber`, the component separates a rich LiBr solution + (:code:`in1`) into a saturated poor solution (:code:`out1`) and water + vapour (:code:`out3`). The heat of desorption is supplied by a heating + fluid (:code:`in2` to :code:`out2`) via a mandatory energy balance, so + no external heat flow specification is required. + + Ports + ----- + - :code:`in1` - rich LiBr solution (SolutionConnection) + - :code:`in2` - heating fluid (Connection) + - :code:`out1` - poor LiBr solution (SolutionConnection) + - :code:`out2` - heating fluid (Connection) + - :code:`out3` - water vapour (Connection) + + .. note:: + + The port numbering deviates from :class:`Desorber`: the heating + fluid occupies :code:`in2` and :code:`out2` in analogy to the + two-sided heat exchangers (matching :code:`pr2` and :code:`dp2`), + therefore the water vapour outlet moves to :code:`out3`. + + Mandatory Equations + ------------------- + - sorption side mass balance: + :math:`\dot m_{in,1} = \dot m_{out,1} + \dot m_{out,3}` + - heating fluid mass flow and fluid composition equality + - LiBr balance: + :math:`\dot m_{in,1} \xi_{in,1} = \dot m_{out,1} \xi_{out,1}` + - pressure equality of the sorption side ports + - saturation: + :math:`\xi_{out,1} = \xi_\text{sat}(p_{out,1},\, T_{out,1})` + - energy balance: + :math:`0 = \sum \dot m_{out} h_{out} - \sum \dot m_{in} h_{in}` + + Parameters + ---------- + Q : float + Heat transferred from the heating fluid (hot) side to the sorption + process (W); always negative. + + ttd_u : float + Upper terminal temperature difference + :math:`T_{in,2} - T_{out,1}` (K). + + ttd_l : float + Lower terminal temperature difference + :math:`T_{out,2} - T_{in,1}` (K). + + pr2 : float + Heating fluid side outlet to inlet pressure ratio. + + dp2 : float + Heating fluid side inlet to outlet absolute pressure change. + """ + + _is_wrapper_branch_source = True + _htf_dh_sign = -1 + + @staticmethod + def inlets(): + return ["in1", "in2"] + + @staticmethod + def outlets(): + return ["out1", "out2", "out3"] + + def _sorption_inlets(self): + return [self.inl[0]] + + def _sorption_outlets(self): + return [self.outl[0], self.outl[2]] + + def _solution_inlet(self): + return self.inl[0] + + def _solution_outlet(self): + return self.outl[0] + + def heat_func(self): + i = self.inl[1] + o = self.outl[1] + return self.Q.val_SI - i.m.val_SI * (o.h.val_SI - i.h.val_SI) + + def calc_Q(self): + i = self.inl[1] + o = self.outl[1] + return i.m.val_SI * (o.h.val_SI - i.h.val_SI) + + def ttd_u_func(self): + T_heat_in = self.inl[1].calc_T() + T_sol_out = self.outl[0].calc_T() + return self.ttd_u.val_SI - T_heat_in + T_sol_out + + def ttd_u_dependents(self): + sol_out = self.outl[0] + return { + "scalars": [[ + self.inl[1].p, self.inl[1].h, sol_out.p, sol_out.h + ]], + "vectors": [{sol_out.fluid: {_LIBR_KEY} & sol_out.fluid.is_var}], + } + + def ttd_l_func(self): + T_heat_out = self.outl[1].calc_T() + T_sol_in = self.inl[0].calc_T() + return self.ttd_l.val_SI - T_heat_out + T_sol_in + + def ttd_l_dependents(self): + sol_in = self.inl[0] + return { + "scalars": [[ + self.outl[1].p, self.outl[1].h, sol_in.p, sol_in.h + ]], + "vectors": [{sol_in.fluid: {_LIBR_KEY} & sol_in.fluid.is_var}], + } + + def _initial_temperature_edges(self): + sorption_ports = self._sorption_inlets() + self._sorption_outlets() + first = sorption_ports[0] + edges = [(first, c, 0.0, 1.0) for c in sorption_ports[1:]] + edges += [(self.inl[1], self.outl[1], 0.0, 1.0)] + ttd_upper = self.ttd_u.val_SI if self.ttd_u.is_set else 10.0 + weight_upper = 5.0 if self.ttd_u.is_set else 0.3 + ttd_lower = self.ttd_l.val_SI if self.ttd_l.is_set else 10.0 + weight_lower = 5.0 if self.ttd_l.is_set else 0.3 + edges += [ + (self.outl[0], self.inl[1], ttd_upper, weight_upper), + (self.inl[0], self.outl[1], ttd_lower, weight_lower), + ] + return edges + + def calc_parameters(self): + super().calc_parameters() + self.ttd_u.val_SI = self.inl[1].T.val_SI - self.outl[0].T.val_SI + self.ttd_l.val_SI = self.outl[1].T.val_SI - self.inl[0].T.val_SI + + def convergence_check(self): + super().convergence_check() + vap_out = self.outl[2] + if not vap_out.h.is_var: + return + try: + from tespy.tools.fluid_properties.functions import h_mix_pQ + from tespy.tools.fluid_properties.functions import phase_mix_ph + phase = phase_mix_ph(vap_out.p.val_SI, vap_out.h.val_SI, vap_out.fluid_data) + if phase != "g": + h_g = h_mix_pQ(vap_out.p.val_SI, 1.0, vap_out.fluid_data) + vap_out.h.set_reference_val_SI(h_g) + except Exception: + pass + + def start_fluid_wrapper_branch(self): + vap_conn = self.outl[2] + vap_branch = { + "connections": [vap_conn], + "components": [self], + } + vap_conn.target.propagate_wrapper_to_target(vap_branch) + + sol_conn = self.outl[0] + sol_branch = { + "connections": [sol_conn], + "components": [self], + } + sol_conn.target.propagate_wrapper_to_target(sol_branch) + + return {vap_conn.label: vap_branch, sol_conn.label: sol_branch} + + def propagate_wrapper_to_target(self, branch): + inconn = branch["connections"][-1] + if inconn == self.inl[1]: + outconn = self.outl[1] + branch["connections"] += [outconn] + branch["components"] += [self] + outconn.target.propagate_wrapper_to_target(branch) + return + if self in branch["components"]: + return + outconn = self.outl[0] + branch["connections"] += [outconn] + branch["components"] += [self] + outconn.target.propagate_wrapper_to_target(branch) diff --git a/src/tespy/connections/__init__.py b/src/tespy/connections/__init__.py index fb582ff1..00e04b8e 100644 --- a/src/tespy/connections/__init__.py +++ b/src/tespy/connections/__init__.py @@ -5,3 +5,4 @@ from .heatconnection import HeatConnection # noqa: F401 from .humidairconnection import HAConnection # noqa: F401 from .powerconnection import PowerConnection # noqa: F401 +from .solutionconnection import SolutionConnection # noqa: F401 diff --git a/src/tespy/connections/solutionconnection.py b/src/tespy/connections/solutionconnection.py new file mode 100644 index 00000000..f7a667dd --- /dev/null +++ b/src/tespy/connections/solutionconnection.py @@ -0,0 +1,172 @@ +# -*- coding: utf-8 +"""Module for the SolutionConnection class. + + +This file is part of project TESPy (github.com/oemof/tespy). It's copyrighted +by the contributors recorded in the version control history of the file, +available from its original location tespy/connections/solutionconnection.py + +SPDX-License-Identifier: MIT +""" +from tespy.tools.data_containers import FluidProperties as dc_prop +from tespy.tools.fluid_properties.functions import h_mix_pT +from tespy.tools.fluid_properties.mixtures import _get_fluid_alias +from tespy.tools.helpers import seeded_random + +from .connection import Connection +from .connection import connection_registry + + +@connection_registry +class SolutionConnection(Connection): + """Connection for binary LiBr-water absorption-cycle streams. + + Locks the mixing rule to :code:`"libr_water"`, which evaluates + thermodynamic properties via CoolProp's :code:`INCOMP::LiBr` backend + (Patek-Klomfar correlations) as a function of temperature and LiBr mass + fraction. + + The LiBr mass fraction :code:`xi` can be set directly as a convenience + parameter; it maps internally to + :code:`fluid={"INCOMP::LiBr": xi, "H2O": 1 - xi}`. + + Notes + ----- + - Only the liquid-solution side is modelled here. The refrigerant + (steam) leaving or entering the absorber/desorber uses a plain + :class:`.Connection` with :code:`fluid={"H2O": 1}`. + - Specifying :code:`fluid={"LiBr": xi, "H2O": 1 - xi}` (without the + :code:`INCOMP::` prefix) is also accepted; the prefix is injected + automatically. + """ + + def get_parameters(self): + params = super().get_parameters() + params["xi"] = dc_prop( + quantity="ratio", + description="LiBr mass fraction in solution (convenience parameter)" + ) + return params + + def _get_mixing_rule(self): + return "libr_water" + + def _set_mixing_rule(self, value): + if value is not None and value != self.mixing_rule: + msg = ( + "You cannot change the mixing rule specification for a " + f"Connection of type {self.__class__.__name__}." + ) + raise ValueError(msg) + + mixing_rule = property(_get_mixing_rule, _set_mixing_rule) + + def _parameter_specification(self, key, value): + if key in ("xi", "xi0"): + if value is None: + self.fluid.is_set = set() + else: + fluid_spec = {"INCOMP::LiBr": value, "H2O": 1.0 - value} + if key == "xi": + self.set_attr(fluid=fluid_spec) + else: + self.set_attr(fluid0=fluid_spec) + else: + super()._parameter_specification(key, value) + + def _fluid_specification(self, key, value): + if key == "fluid" and isinstance(value, dict): + translated = {} + for fluid_name, fraction in value.items(): + if "::" not in fluid_name and fluid_name.lower() == "libr": + fluid_name = f"INCOMP::{fluid_name}" + translated[fluid_name] = fraction + value = translated + super()._fluid_specification(key, value) + + def _presolve(self): + water_alias = _get_fluid_alias("H2O", self.fluid_data) + if not water_alias: + msg = ( + f"H2O must be present in the fluid composition of " + f"SolutionConnection {self.label!r}." + ) + raise ValueError(msg) + + if len(self.fluid.is_var) > 0: + return [] + + presolved_equations = [] + if self.h.is_var and not self.p.is_var and self.T.is_set: + self.h.set_reference_val_SI( + h_mix_pT(self.p.val_SI, self.T.val_SI, self.fluid_data, self.mixing_rule) + ) + self.h._potential_var = False + if "T" in self._equation_set_lookup.values(): + presolved_equations += ["T"] + + return [ + key + for parameter in presolved_equations + for key, val in self._equation_set_lookup.items() + if val == parameter + ] + + def _precalc_guess_values(self): + if not self.h.is_var: + return False + if not self.good_starting_values and self.T.is_set: + try: + self.h.set_reference_val_SI( + h_mix_pT(self.p.val_SI, self.T.val_SI, self.fluid_data, self.mixing_rule) + ) + return True + except Exception: + pass + return False + + def _adjust_to_property_limits(self, nw): + ref = self.fluid._reference_container + if ref is not None and "LiBr" in self.fluid.is_var: + xi = self.fluid.val.get("LiBr", 0) + if xi > 0.74 or xi < 0.01: + xi_clipped = min(max(xi, 0.01), 0.74) + ref.val["LiBr"] = xi_clipped + ref.val["H2O"] = 1.0 - xi_clipped + super()._adjust_to_property_limits(nw) + + def _guess_starting_values(self, units, covered): + h_sources = super()._guess_starting_values(units, covered) + if self.h.is_var and not self.good_starting_values: + reference = self.h._reference_container + if reference not in covered: + rand = seeded_random(self.label) + T_rand = 310 + rand * (420 - 310) + try: + h = h_mix_pT(1e5, T_rand, self.fluid_data, self.mixing_rule) + self.h.set_reference_val_SI(h) + covered.add(reference) + h_sources.append(reference) + except Exception: + pass + return h_sources + + def calc_xi(self): + """Return solved LiBr mass fraction from fluid composition.""" + water = _get_fluid_alias("H2O", self.fluid_data) + if water: + return 1.0 - self.fluid.val[next(iter(water))] + return sum(self.fluid.val.values()) + + def calc_results(self, units, skip_postprocess): + if not skip_postprocess: + self.xi.val_SI = self.calc_xi() + return super().calc_results(units, skip_postprocess) + + @classmethod + def _result_attributes(cls): + return ["m", "p", "h", "T", "v", "s", "vol", "xi"] + + @classmethod + def _print_attributes(cls): + return ["m", "p", "h", "T", "xi"] diff --git a/src/tespy/tools/fluid_properties/functions.py b/src/tespy/tools/fluid_properties/functions.py index 9d4d268e..bb0425c2 100644 --- a/src/tespy/tools/fluid_properties/functions.py +++ b/src/tespy/tools/fluid_properties/functions.py @@ -19,6 +19,8 @@ from .helpers import get_pure_fluid from .helpers import inverse_temperature_mixture from .mixtures import MIXING_RULES +from .mixtures import T_mix_ph_libr_water +from .mixtures import phase_mix_ph_libr_water from .mixtures import w_mix_ph_humidair from .mixtures import w_mix_ps_humidair @@ -261,6 +263,8 @@ def T_mix_ph(p, h, fluid_data, mixing_rule=None, T0=None): if mixing_rule == "humidair": w = w_mix_ph_humidair(p, h, fluid_data) return HAPropsSI("T", "P", p, "H", h, "W", w) + elif mixing_rule == "libr_water": + return T_mix_ph_libr_water(p, h, fluid_data, T0) else: kwargs = { "p": p, "target_value": h, "fluid_data": fluid_data, "T0": T0, @@ -602,6 +606,8 @@ def phase_mix_ph(p, h, fluid_data, mixing_rule=None): :code:`mixing_rule` is not recognised. """ if get_number_of_fluids(fluid_data) != 1: + if mixing_rule == "libr_water": + return phase_mix_ph_libr_water(p, h, fluid_data) if mixing_rule not in _MIXING_RULE_PHASE: raise ValueError( f"Cannot determine phase for multi-component fluid data with " diff --git a/src/tespy/tools/fluid_properties/mixtures.py b/src/tespy/tools/fluid_properties/mixtures.py index 04abfaf7..09fe61a0 100644 --- a/src/tespy/tools/fluid_properties/mixtures.py +++ b/src/tespy/tools/fluid_properties/mixtures.py @@ -11,6 +11,7 @@ SPDX-License-Identifier: MIT """ +import CoolProp as _CP from CoolProp.CoolProp import HAPropsSI from tespy.tools.global_vars import FLUID_ALIASES @@ -1149,6 +1150,287 @@ def T_ph(self, name): def T_ps(self, name): return self._get(self._T_ps, name, "temperature (from entropy)") +_LIBR_PROPS_CACHE = (None, None, None, None) # (p, T, xi, (h, s, v)) +_LIBR_T_EPS = 0.01 + + +def _libr_wrapper(fluid_data): + water = _get_fluid_alias("H2O", fluid_data) + for fluid, data in fluid_data.items(): + if fluid not in water and _is_larger_than_precision(data["mass_fraction"]): + return data["wrapper"] + return None + + +def _xi_libr(fluid_data): + water = _get_fluid_alias("H2O", fluid_data) + if water: + return 1.0 - fluid_data[next(iter(water))]["mass_fraction"] + return sum(d["mass_fraction"] for d in fluid_data.values()) + + +def _xi_sat_libr(p, T, fluid_data): + r"""LiBr mass fraction at which :math:`p_\text{sat}(T, \xi) = p`. + + Uses :func:`scipy.optimize.brentq` over :math:`\xi \in [0.001, 0.749]`. + :math:`p_\text{sat}` is monotonically decreasing in :math:`\xi` (higher + LiBr concentration lowers the water vapour pressure). + """ + from scipy.optimize import brentq + + libr_as = _libr_wrapper(fluid_data).AS + + def residual(xi): + libr_as.set_mass_fractions([xi]) + libr_as.update(_CP.QT_INPUTS, 0, T) + return libr_as.p() - p + + r_lo = residual(0.001) + if r_lo <= 0: + return 0.001 + r_hi = residual(0.749) + if r_hi >= 0: + return 0.749 + return brentq(residual, 0.001, 0.749, xtol=1e-6) + + +def _T_sat_libr(p, xi, fluid_data): + r"""Saturation temperature of LiBr-H2O at pressure *p* and mass fraction *xi*. + + Inverts :func:`_xi_sat_libr` over temperature using + :func:`scipy.optimize.brentq`. :math:`p_\text{sat}(T, \xi)` is + monotonically increasing in *T*. + """ + from scipy.optimize import brentq + + w = _libr_wrapper(fluid_data) + xi_safe = max(0.001, min(0.749, xi)) + libr_as = w.AS + + def residual(T): + libr_as.set_mass_fractions([xi_safe]) + libr_as.update(_CP.QT_INPUTS, 0, T) + return libr_as.p() - p + + T_min = w._T_min + 2 * _LIBR_T_EPS + T_max = w._T_max - 2 * _LIBR_T_EPS + if residual(T_min) >= 0: + return T_min + if residual(T_max) <= 0: + return T_max + return brentq(residual, T_min, T_max, xtol=1e-6) + + +def _p_sat_libr(T, xi, fluid_data): + r"""Vapour pressure of LiBr-H2O at temperature *T* and mass fraction *xi*.""" + xi_safe = max(0.001, min(0.749, xi)) + libr_as = _libr_wrapper(fluid_data).AS + libr_as.set_mass_fractions([xi_safe]) + libr_as.update(_CP.QT_INPUTS, 0, T) + return libr_as.p() + + +def _libr_props_compute(p, T, xi, fluid_data): + r"""Compute :math:`(h, s, v)` for LiBr-H2O at *(p, T, xi)*. + + When :math:`p \geq p_\text{sat}(T, \xi)` the solution is fully liquid and + properties are read directly from CoolProp's :code:`INCOMP::LiBr` backend. + + When :math:`p < p_\text{sat}(T, \xi)` the solution is in equilibrium with + water vapour. The saturated LiBr fraction :math:`\xi_\text{sat}` is found + via :func:`_xi_sat_libr` such that :math:`p_\text{sat}(T, \xi_\text{sat}) = p`. + Per-unit-mass balances give liquid fraction + :math:`m_l = \xi / \xi_\text{sat}` and vapour fraction + :math:`m_v = 1 - m_l`. Properties are the weighted sum of the saturated + liquid solution and pure water vapour. + + """ + libr_as = _libr_wrapper(fluid_data).AS + libr_as.set_mass_fractions([xi]) + libr_as.update(_CP.QT_INPUTS, 0, T) + p_sat = libr_as.p() + + if p >= p_sat: + libr_as.update(_CP.PT_INPUTS, p, T) + return libr_as.hmass(), libr_as.smass(), 1.0 / libr_as.rhomass() + + xi_sat = _xi_sat_libr(p, T, fluid_data) + m_l = xi / xi_sat + m_v = 1.0 - m_l + + libr_as.set_mass_fractions([xi_sat]) + libr_as.update(_CP.QT_INPUTS, 0, T) + p_sat_new = libr_as.p() + libr_as.update(_CP.PT_INPUTS, p_sat_new + 1.0, T) + h_l = libr_as.hmass() + s_l = libr_as.smass() + v_l = 1.0 / libr_as.rhomass() + + water = _get_fluid_alias("H2O", fluid_data) + w = fluid_data[next(iter(water))]["wrapper"] + h_v = w.h_QT(1, T) + s_v = w.s_QT(1, T) + v_v = 1.0 / w.d_QT(1, T) + + h0 = m_l * h_l + m_v * h_v + s0 = m_l * s_l + m_v * s_v + v0 = m_l * v_l + m_v * v_v + return h0, s0, v0 + + +def _get_libr_props(p, T, fluid_data): + global _LIBR_PROPS_CACHE + xi = _xi_libr(fluid_data) + cp, cT, cxi, cached = _LIBR_PROPS_CACHE + if p == cp and T == cT and xi == cxi: + return cached + props = _libr_props_compute(p, T, xi, fluid_data) + _LIBR_PROPS_CACHE = (p, T, xi, props) + return props + + +def h_mix_pT_libr_water(p, T, fluid_data, **kwargs): + r""" + Calculate specific enthalpy of a LiBr-water solution. + + Uses CoolProp's :code:`INCOMP::LiBr` backend (Patek-Klomfar correlations). + Handles both the sub-saturated liquid and the two-phase (solution + water + vapour) region; see :func:`_libr_props_compute` for details. + + Parameters + ---------- + p : float + Pressure in Pa. + T : float + Temperature in K. + fluid_data : dict + Fluid property data containing LiBr and H2O components. + **kwargs + Ignored; present for interface compatibility. + + Returns + ------- + float + Specific enthalpy in J/kg. + """ + return _get_libr_props(p, T, fluid_data)[0] + + +def s_mix_pT_libr_water(p, T, fluid_data, **kwargs): + r""" + Calculate specific entropy of a LiBr-water solution. + + Uses CoolProp's :code:`INCOMP::LiBr` backend. Handles both the + sub-saturated liquid and the two-phase region; see + :func:`_libr_props_compute` for details. + + Parameters + ---------- + p : float + Pressure in Pa. + T : float + Temperature in K. + fluid_data : dict + Fluid property data containing LiBr and H2O components. + **kwargs + Ignored; present for interface compatibility. + + Returns + ------- + float + Specific entropy in J/(kg K). + """ + return _get_libr_props(p, T, fluid_data)[1] + + +def v_mix_pT_libr_water(p, T, fluid_data, **kwargs): + r""" + Calculate specific volume of a LiBr-water solution. + + Uses CoolProp's :code:`INCOMP::LiBr` backend. Handles both the + sub-saturated liquid and the two-phase region; see + :func:`_libr_props_compute` for details. + + Parameters + ---------- + p : float + Pressure in Pa. + T : float + Temperature in K. + fluid_data : dict + Fluid property data containing LiBr and H2O components. + **kwargs + Ignored; present for interface compatibility. + + Returns + ------- + float + Specific volume in m³/kg. + """ + return _get_libr_props(p, T, fluid_data)[2] + + +def phase_mix_ph_libr_water(p, h, fluid_data): + r"""Return the phase of a LiBr-water stream given *(p, h)*. + + Returns :code:`"l"` when the solution is fully liquid + (:math:`p \geq p_\text{sat}(T, \xi)`) and :code:`"tp"` when the system + is in the two-phase (solution + water vapour) region. + """ + T = T_mix_ph_libr_water(p, h, fluid_data) + xi = _xi_libr(fluid_data) + libr_as = _libr_wrapper(fluid_data).AS + libr_as.set_mass_fractions([xi]) + libr_as.update(_CP.QT_INPUTS, 0, T) + return "l" if p >= libr_as.p() else "tp" + + +def T_mix_ph_libr_water(p, h, fluid_data, T0=None): + r""" + Invert :func:`h_mix_pT_libr_water` to recover temperature. + + Uses :func:`scipy.optimize.brentq` over the valid temperature range of + the LiBr wrapper, inset by :code:`2 * _LIBR_T_EPS` from each boundary. + This avoids Newton convergence failures near the liquid-to-two-phase + boundary where :math:`dh/dT` changes sharply. + + Parameters + ---------- + p : float + Pressure in Pa. + h : float + Specific enthalpy in J/kg. + fluid_data : dict + Fluid property data containing LiBr and H2O components. + T0 : float, optional + Ignored; present for interface compatibility. + + Returns + ------- + float + Temperature in K. + """ + from scipy.optimize import brentq + + xi = _xi_libr(fluid_data) + w = _libr_wrapper(fluid_data) + T_lo = w._T_min + 2 * _LIBR_T_EPS + T_hi = w._T_max - 2 * _LIBR_T_EPS + + def residual(T): + return _libr_props_compute(p, T, xi, fluid_data)[0] - h + + # clamp value to be within T_lo/T_hi to prevent inter-iteration errors + h_lo = _libr_props_compute(p, T_lo, xi, fluid_data)[0] + if h <= h_lo: + return T_lo + h_hi = _libr_props_compute(p, T_hi, xi, fluid_data)[0] + if h >= h_hi: + return T_hi + + return brentq(residual, T_lo, T_hi, xtol=1e-6) + + MIXING_RULES = MixingRuleRegistry() MIXING_RULES.register( @@ -1184,3 +1466,9 @@ def T_ps(self, name): v_pT=v_mix_pT_humidair, viscosity_pT=viscosity_mix_pT_humidair, ) +MIXING_RULES.register( + "libr_water", + h_pT=h_mix_pT_libr_water, + s_pT=s_mix_pT_libr_water, + v_pT=v_mix_pT_libr_water, +) diff --git a/tests/test_connections.py b/tests/test_connections.py index 288c23fa..acdf658e 100644 --- a/tests/test_connections.py +++ b/tests/test_connections.py @@ -26,6 +26,7 @@ from tespy.connections import Connection from tespy.connections import HeatConnection from tespy.connections import PowerConnection +from tespy.connections import SolutionConnection from tespy.connections import Ref from tespy.connections.connection import ConnectionBase from tespy.connections.connection import connection_registry @@ -579,7 +580,7 @@ def test_all_classes_in_registry(obj): def make_connection(cls): - if cls == Connection or cls == HAConnection: + if cls == Connection or cls == HAConnection or cls == SolutionConnection: return cls(Source(""), "out1", Sink(""), "in1") elif cls == PowerConnection: return cls(PowerSource(""), "power", PowerSink(""), "power")