Ates multiport

CHANGELOG.md

@@ -1,13 +1,14 @@
-# [Unreleased-main] - 2025-06-25
+# [Unreleased-main] - 2025-08-15
 
 ## Added
-- xx
+- ATES multiport asset, which contains functionalities that can only be applied with user-defined constraints.
+- ATES_MULTI_PORT_NOT_SUPPORTED error that ensures that the ATES multiport asset is not used in existing workflows as under development.
 
 ## Changed
 - Previously variable operational cost of air-to-water heat pump was based on  the thermal power usage. Now it is based on the electrical power usage
 
 ## Fixed
-- xxx
+- Check if there are no varying temperatures used to allow for updating pipe diameter bounds connected to demands and producers.
 
 
 # [0.1.13] - 2025-06-25

src/mesido/component_type_mixin.py

@@ -301,49 +301,104 @@ def pre(self):
         ates_connections = {}
 
         for a in atess:
-            ates_connections[a] = []
+            if a not in self.energy_system_components.get("ates_multi_port", []):
+                ates_connections[a] = []
 
-            for k in ["In", "Out"]:
-                a_conn = f"{a}.{heat_network_model_type}{k}"
-                prop = (
-                    "T" if heat_network_model_type == "QTH" else NetworkSettings.NETWORK_TYPE_HEAT
-                )
-                aliases = [
-                    x
-                    for x in self.alias_relation.aliases(f"{a_conn}.{prop}")
-                    if not x.startswith(a) and x.endswith(f".{prop}")
-                ]
+                for k in ["In", "Out"]:
+                    a_conn = f"{a}.{heat_network_model_type}{k}"
+                    prop = (
+                        "T"
+                        if heat_network_model_type == "QTH"
+                        else NetworkSettings.NETWORK_TYPE_HEAT
+                    )
+                    aliases = [
+                        x
+                        for x in self.alias_relation.aliases(f"{a_conn}.{prop}")
+                        if not x.startswith(a) and x.endswith(f".{prop}")
+                    ]
+
+                    if len(aliases) > 1:
+                        raise Exception(f"More than one connection to {a_conn}")
+                    elif len(aliases) == 0:
+                        continue
+                    #     raise Exception(f"Found no connection to {a_conn}")
 
-                if len(aliases) > 1:
-                    raise Exception(f"More than one connection to {a_conn}")
-                elif len(aliases) == 0:
-                    raise Exception(f"Found no connection to {a_conn}")
+                    in_suffix = ".QTHIn.T" if heat_network_model_type == "QTH" else ".HeatIn.Heat"
+                    out_suffix = (
+                        ".QTHOut.T" if heat_network_model_type == "QTH" else ".HeatOut.Heat"
+                    )
 
-                in_suffix = ".QTHIn.T" if heat_network_model_type == "QTH" else ".HeatIn.Heat"
-                out_suffix = ".QTHOut.T" if heat_network_model_type == "QTH" else ".HeatOut.Heat"
+                    if aliases[0].endswith(out_suffix):
+                        asset_w_orientation = (
+                            aliases[0][: -len(out_suffix)],
+                            NodeConnectionDirection.IN,
+                        )
+                    else:
+                        assert aliases[0].endswith(in_suffix)
+                        asset_w_orientation = (
+                            aliases[0][: -len(in_suffix)],
+                            NodeConnectionDirection.OUT,
+                        )
 
-                if aliases[0].endswith(out_suffix):
-                    asset_w_orientation = (
-                        aliases[0][: -len(out_suffix)],
-                        NodeConnectionDirection.IN,
+                    assert asset_w_orientation[0] in pipes_set
+
+                    if k == "Out":
+                        assert self.is_cold_pipe(asset_w_orientation[0])
+                    else:
+                        assert self.is_hot_pipe(asset_w_orientation[0])
+
+                    ates_connections[a].append(asset_w_orientation)
+
+                ates_connections[a] = tuple(ates_connections[a])
+            else:
+                ates_connections[a] = []
+
+                for k in ["In", "Out"]:
+                    a_conn = f"{a}.ChargeHot.{heat_network_model_type}{k}"
+                    prop = (
+                        "T"
+                        if heat_network_model_type == "QTH"
+                        else NetworkSettings.NETWORK_TYPE_HEAT
                     )
-                else:
-                    assert aliases[0].endswith(in_suffix)
-                    asset_w_orientation = (
-                        aliases[0][: -len(in_suffix)],
-                        NodeConnectionDirection.OUT,
+                    aliases = [
+                        x
+                        for x in self.alias_relation.aliases(f"{a_conn}.{prop}")
+                        if not x.startswith(a) and x.endswith(f".{prop}")
+                    ]
+
+                    if len(aliases) > 1:
+                        raise Exception(f"More than one connection to {a_conn}")
+                    elif len(aliases) == 0:
+                        continue
+                    #     raise Exception(f"Found no connection to {a_conn}")
+
+                    in_suffix = ".QTHIn.T" if heat_network_model_type == "QTH" else ".HeatIn.Heat"
+                    out_suffix = (
+                        ".QTHOut.T" if heat_network_model_type == "QTH" else ".HeatOut.Heat"
                     )
 
-                assert asset_w_orientation[0] in pipes_set
+                    if aliases[0].endswith(out_suffix):
+                        asset_w_orientation = (
+                            aliases[0][: -len(out_suffix)],
+                            NodeConnectionDirection.IN,
+                        )
+                    else:
+                        assert aliases[0].endswith(in_suffix)
+                        asset_w_orientation = (
+                            aliases[0][: -len(in_suffix)],
+                            NodeConnectionDirection.OUT,
+                        )
 
-                if k == "Out":
-                    assert self.is_cold_pipe(asset_w_orientation[0])
-                else:
-                    assert self.is_hot_pipe(asset_w_orientation[0])
+                    assert asset_w_orientation[0] in pipes_set
+
+                    # if k == "Out":
+                    #     assert self.is_cold_pipe(asset_w_orientation[0])
+                    # else:
+                    #     assert self.is_hot_pipe(asset_w_orientation[0])
 
-                ates_connections[a].append(asset_w_orientation)
+                    ates_connections[a].append(asset_w_orientation)
 
-            ates_connections[a] = tuple(ates_connections[a])
+                ates_connections[a] = tuple(ates_connections[a])
 
         demand_connections = {}
 

src/mesido/esdl/asset_to_component_base.py

@@ -233,6 +233,11 @@ class _AssetToComponentBase:
     primary_port_name_convention = "primary"
     secondary_port_name_convention = "secondary"
 
+    charge_port_name_convention = "charge"
+    discharge_port_name_convention = "discharge"
+    hot_port_name_convention = "hot"
+    cold_port_name_convention = "cold"
+
     __power_keys = ["power", "maxDischargeRate", "maxChargeRate", "capacity"]
 
     def __init__(self, **kwargs):
@@ -982,7 +987,7 @@ def _get_connected_q_nominal(self, asset: Asset) -> Union[float, Dict]:
             if q_nominals["Q_nominal"] is not None:
                 self._port_to_q_nominal[asset.out_ports[0]] = q_nominals["Q_nominal"]
                 return q_nominals
-        elif len(asset.in_ports) >= 2 and len(asset.out_ports) == 2:
+        elif len(asset.in_ports) >= 2 and len(asset.out_ports) >= 2:
             q_nominals = {}
             for p in asset.in_ports:
                 if isinstance(p.carrier, esdl.HeatCommodity):
@@ -1010,10 +1015,29 @@ def _get_connected_q_nominal(self, asset: Asset) -> Union[float, Dict]:
                     if q_nominal is not None:
                         self._port_to_q_nominal[p] = q_nominal
                         self._port_to_q_nominal[out_port] = q_nominal
-                        if "sec" in p.name.lower():
+                        if self.secondary_port_name_convention in p.name.lower():
                             q_nominals["Secondary"] = {"Q_nominal": q_nominal}
-                        else:
+                        elif self.primary_port_name_convention in p.name.lower():
                             q_nominals["Primary"] = {"Q_nominal": q_nominal}
+                        elif (
+                            self.discharge_port_name_convention in p.name.lower()
+                            and self.hot_port_name_convention in p.name.lower()
+                        ):
+                            q_nominals["DischargeHot"] = {"Q_nominal": q_nominal}
+                        elif (
+                            self.discharge_port_name_convention in p.name.lower()
+                            and self.cold_port_name_convention in p.name.lower()
+                        ):
+                            q_nominals["DischargeCold"] = {"Q_nominal": q_nominal}
+                        elif (
+                            self.charge_port_name_convention in p.name.lower()
+                            and self.hot_port_name_convention in p.name.lower()
+                        ):
+                            q_nominals["ChargeHot"] = {"Q_nominal": q_nominal}
+                        else:
+                            raise Exception(
+                                f"{asset.name} has ports that do not comply with any " f"convention"
+                            )
 
             return q_nominals
 
@@ -1220,6 +1244,75 @@ def _supply_return_temperature_modifiers(self, asset: Asset) -> MODIFIERS:
                 },
             }
             return temperatures
+        elif len(asset.in_ports) == 3 and len(asset.out_ports) == 3 and asset.asset_type == "ATES":
+            prim_return_temperature = None
+            sec_return_temperature = None
+            for p in asset.in_ports:
+                carrier = asset.global_properties["carriers"][p.carrier.id]
+                if (
+                    self.discharge_port_name_convention in p.name.lower()
+                    and self.hot_port_name_convention in p.name.lower()
+                ):
+                    discharge_hot_in_temperature_id = carrier["id_number_mapping"]
+                    discharge_hot_in_temperature = carrier["temperature"]
+                elif (
+                    self.discharge_port_name_convention in p.name.lower()
+                    and self.cold_port_name_convention in p.name.lower()
+                ):
+                    discharge_cold_in_temperature_id = carrier["id_number_mapping"]
+                    discharge_cold_in_temperature = carrier["temperature"]
+                elif (
+                    self.charge_port_name_convention in p.name.lower()
+                    and self.hot_port_name_convention in p.name.lower()
+                ):
+                    charge_hot_in_temperature_id = carrier["id_number_mapping"]
+                    charge_hot_in_temperature = carrier["temperature"]
+                else:
+                    raise RuntimeError
+            for p in asset.out_ports:
+                carrier = asset.global_properties["carriers"][p.carrier.id]
+                if (
+                    self.discharge_port_name_convention in p.name.lower()
+                    and self.hot_port_name_convention in p.name.lower()
+                ):
+                    discharge_hot_out_temperature_id = carrier["id_number_mapping"]
+                    discharge_hot_out_temperature = carrier["temperature"]
+                elif (
+                    self.discharge_port_name_convention in p.name.lower()
+                    and self.cold_port_name_convention in p.name.lower()
+                ):
+                    discharge_cold_out_temperature_id = carrier["id_number_mapping"]
+                    discharge_cold_out_temperature = carrier["temperature"]
+                elif (
+                    self.charge_port_name_convention in p.name.lower()
+                    and self.hot_port_name_convention in p.name.lower()
+                ):
+                    charge_hot_out_temperature_id = carrier["id_number_mapping"]
+                    charge_hot_out_temperature = carrier["temperature"]
+                else:
+                    raise RuntimeError
+            temperatures = {
+                "DischargeHot": {
+                    "T_supply": discharge_hot_out_temperature,
+                    "T_return": discharge_hot_in_temperature,
+                    "T_supply_id": discharge_hot_out_temperature_id,
+                    "T_return_id": discharge_hot_in_temperature_id,
+                },
+                "DischargeCold": {
+                    "T_supply": discharge_cold_out_temperature,
+                    "T_return": discharge_cold_in_temperature,
+                    "T_supply_id": discharge_cold_out_temperature_id,
+                    "T_return_id": discharge_cold_in_temperature_id,
+                },
+                "ChargeHot": {
+                    "T_supply": charge_hot_in_temperature,
+                    "T_return": charge_hot_out_temperature,
+                    "T_supply_id": charge_hot_in_temperature_id,
+                    "T_return_id": charge_hot_out_temperature_id,
+                },
+            }
+            return temperatures
+
         else:
             # unknown model type
             return {}

src/mesido/esdl/esdl_additional_vars_mixin.py

@@ -33,7 +33,13 @@ def read(self):
         # ensure that we don't have unneeded large amount of available pipe classes for pipes
         # connected to smaller demands.
         # TODO: add the same for electricity ones we have proper support for that in the ESDLMixin
-        if len(self.temperature_carriers().items()) == 0:
+        varying_temperatures = False
+        for _carrier, temperatures in self.temperature_carriers().items():
+            carrier_id_number_mapping = str(temperatures["id_number_mapping"])
+            temperature_regimes = self.temperature_regimes(int(carrier_id_number_mapping))
+            varying_temperatures = varying_temperatures if len(temperature_regimes) <= 1 else True
+
+        if not varying_temperatures:
             for asset, (
                 connected_asset,
                 _orientation,