Bugfix to technology_graph and other minor fixes

CHANGELOG.md

@@ -78,6 +78,7 @@
 - Corrected timestamps in OpenMeteo resource downloads when resource data is downloaded in local time [PR #814](https://github.com/NatLabRockies/H2Integrate/pull/814)
 - Added a thermal-nuclear (light-water reactor) model and a high-temperature steam electrolysis model. [PR 807](https://github.com/NatLabRockies/H2Integrate/pull/807)
 - Fixed docs build warnings by correcting inline-literal docstring markup etc, also enabled warning-as-error in the shared docs build script to minimize number of future warnings. [PR 821](https://github.com/NatLabRockies/H2Integrate/pull/821)
+- Updated edge attribute `commodity` of in `H2Integrate.create_technology_graph` to use lists instead of strings to account for systems with multiple commodities connected between two technologies [PR 823](https://github.com/NatLabRockies/H2Integrate/pull/823)
 
 ## 0.8 [April 15, 2026]
 

h2integrate/control/control_strategies/system_level/system_level_control_base.py

@@ -854,7 +854,7 @@ def get_upstream_techs_for_commodity(
         ancestors_with_commodity = {
             src
             for src, _, comm in self.technology_graph.edges(data="commodity")
-            if src in ancestors and comm == commodity
+            if src in ancestors and commodity in (comm or [])
         }
 
         # Intersect with controller-managed techs

h2integrate/core/h2integrate_model.py

@@ -568,11 +568,12 @@ def _classify_slc_technologies(self):
             tech for tech in upstream_techs if nx.has_path(self.technology_graph, tech, demand_tech)
         }
 
-        sources_to_commodities = {
-            (e[0], e[-1])
-            for e in self.technology_graph.edges(data="commodity")
-            if (e[-1] is not None) and (e[0] in upstream_controllable_techs)
-        }
+        sources_to_commodities = set()
+        for source, _, commodities in self.technology_graph.edges(data="commodity"):
+            if source not in upstream_controllable_techs or commodities is None:
+                continue
+
+            sources_to_commodities.update((source, commodity) for commodity in commodities)
 
         # re-make technology interconnections using only technologies
         # upstream of the demand component
@@ -581,12 +582,21 @@ def _classify_slc_technologies(self):
             for connection in tech_interconnections
             if connection[0] in upstream_controllable_techs
         ]
+
         upstream_tech_graph = self.create_technology_graph(upstream_interconnections)
         slc_topology["technology_graph"] = upstream_tech_graph
 
+        # downselect the technology control classifiers to only include those upstream
+        # of the demand
+        upstream_tech_control_classifiers = {
+            k: v
+            for k, v in self.tech_control_classifiers.items()
+            if k in upstream_controllable_techs
+        }
+
         # Check if storage models have a controller
         storage_tech_to_control = {}
-        for tech, classifier in self.tech_control_classifiers.items():
+        for tech, classifier in upstream_tech_control_classifiers.items():
             if classifier == "storage":
                 control_model = (
                     self.technology_config["technologies"][tech]
@@ -611,7 +621,7 @@ def _classify_slc_technologies(self):
         tech_to_commodity = {
             (e[0], e[-1])
             for e in sources_to_commodities
-            if self.tech_control_classifiers[e[0]] in control_classifiers_to_connect
+            if upstream_tech_control_classifiers.get(e[0]) in control_classifiers_to_connect
         }
         slc_topology["tech_to_commodity"] = tech_to_commodity
 
@@ -620,7 +630,7 @@ def _classify_slc_technologies(self):
         slc_topology["demand_commodity"] = all_params["commodity"]
         slc_topology["demand_commodity_rate_units"] = all_params.get("commodity_rate_units", None)
 
-        slc_topology["tech_control_classifiers"] = self.tech_control_classifiers
+        slc_topology["tech_control_classifiers"] = upstream_tech_control_classifiers
 
         return slc_topology
 
@@ -2173,11 +2183,37 @@ def create_technology_graph(self, tech_interconnections: list | set):
         """
         technology_graph = nx.DiGraph()
 
+        def _as_commodity_list(commodity):
+            """Coerce a commodity definition to a list."""
+            if commodity is None:
+                return []
+            if isinstance(commodity, str):
+                return [commodity]
+            return list(commodity)
+
         for connection in tech_interconnections:
             source = connection[0]
             destination = connection[1]
             if len(connection) == 4:
-                technology_graph.add_edge(source, destination, commodity=connection[2])
+                new_commodities = _as_commodity_list(connection[2])
+
+                # Commodity is defined in connection. Keep edge commodities
+                # as a list, even for a single commodity.
+                if technology_graph.has_edge(source, destination):
+                    connected_cmods = technology_graph.edges[source, destination].get("commodity")
+                    existing_commodities = _as_commodity_list(connected_cmods)
+                    merged_commodities = list(set(existing_commodities + new_commodities))
+                    technology_graph.add_edge(
+                        source,
+                        destination,
+                        commodity=merged_commodities,
+                    )
+                else:
+                    technology_graph.add_edge(
+                        source,
+                        destination,
+                        commodity=new_commodities,
+                    )
             else:
                 technology_graph.add_edge(source, destination)
 
@@ -2234,13 +2270,15 @@ def _has_commodity_param(params, commodity, direction):
         # Validate commodity connections
         invalid_outputs = set()  # (tech, commodity) pairs where source lacks _out param
         invalid_inputs = set()  # (tech, commodity) pairs where dest lacks _in param
-        for source, dest, commodity in self.technology_graph.edges(data="commodity"):
-            if commodity is None:
+        for source, dest, commodities in self.technology_graph.edges(data="commodity"):
+            if commodities is None:
                 continue  # length-3 connections have no commodity to check
-            if not _has_commodity_param(tech_io[source], commodity, "out"):
-                invalid_outputs.add((source, commodity))
-            if not _has_commodity_param(tech_io[dest], commodity, "in"):
-                invalid_inputs.add((dest, commodity))
+
+            for commodity in commodities:
+                if not _has_commodity_param(tech_io[source], commodity, "out"):
+                    invalid_outputs.add((source, commodity))
+                if not _has_commodity_param(tech_io[dest], commodity, "in"):
+                    invalid_inputs.add((dest, commodity))
 
         # Build a single error message grouping output and input issues separately
         if invalid_outputs or invalid_inputs:
@@ -2260,27 +2298,6 @@ def _has_commodity_param(params, commodity, direction):
             parts.append(f"Update `technology_interconnections` in {self.plant_config_path}.")
             raise ValueError("\n".join(parts))
 
-    def _get_commodity_for_tech(self, tech_name):
-        """Get a list of the commodities produced for a technology.
-
-        Args:
-            tech_name (str): name of technology
-
-        Returns:
-            list[str]: list of commodities produced by the tech_name
-        """
-        # Define the commodities produced by each technology from technology_interconnections
-        # Each element of the set is a tuple of (source_tech, commodity_produced)
-        self.techs_to_commodities = {
-            (e[0], e[-1])
-            for e in self.technology_graph.edges(data="commodity")
-            if e[-1] is not None
-        }
-
-        tech_commodities = [e[1] for e in self.techs_to_commodities if e[0] == tech_name]
-
-        return tech_commodities
-
     @staticmethod
     def _split_indices_from_connected_parameter_definition(connected_parameter):
         """Extract and parse slice indices from connected parameter definitions for OpenMDAO

h2integrate/core/test/inputs/tech_connection_cases.yaml

@@ -6,10 +6,13 @@ control_classifiers:
   grid_buy: dispatchable
   ng_feedstock: feedstock
   n2_feedstock: feedstock
+  ocean: feedstock
   h2_storage: storage
   battery: storage
+  nh3_storage: storage
   electrolyzer: dispatchable
   haber_bosch: dispatchable
+  desalination: dispatchable
 
 
 
@@ -155,3 +158,38 @@ ammonia_system_nh3_dmd_with_upstream_demand:
   system_level_control:
     control_strategy: DemandFollowingControl
     demand_component: nh3_load_demand
+
+fake_complex_system:
+  technology_interconnections:
+    # create a purified water system for the electrolyzer
+    - [ocean, desalination, salt_water, pipe]
+    - [desalination, electrolyzer, fresh_water, pipe]
+    # add ng feedstock to natural gas plant
+    - [ng_feedstock, natural_gas_plant, natural_gas, pipe]
+    # connect renewable system
+    - [wind, combiner, electricity, cable]
+    - [solar, combiner, electricity, cable]
+    - [combiner, battery, electricity, cable]
+    # combine the battery output with the wind and solar generation
+    - [battery, elec_combiner, electricity, cable]
+    - [combiner, elec_combiner, electricity, cable]
+    # connect the electricity supply to the electrolyzer
+    - [elec_combiner, electrolyzer, electricity, cable]
+    # connect some fake outputs from the electrolyzer to the haber_bosch
+    - [electrolyzer, haber_bosch, hydrogen, pipe]
+    - [electrolyzer, haber_bosch, oxygen, pipe]
+    - [electrolyzer, haber_bosch, heat, pipe]
+    # connect other feedstocks to the ammonia plant
+    - [n2_feedstock, haber_bosch, nitrogen, pipe]
+    - [natural_gas_plant, haber_bosch, electricity, cable]
+    # connect haber_bosch to ammonia storage
+    - [haber_bosch, nh3_storage, ammonia, pipe]
+    # combine the ammonia production with ammonia storage
+    - [haber_bosch, ammonia_combiner, ammonia, pipe]
+    - [nh3_storage, ammonia_combiner, ammonia, pipe]
+    # connect the combined ammonia production to the demand
+    - [ammonia_combiner, nh3_load_demand, ammonia, pipe]
+    # etc
+  system_level_control:
+    control_strategy: DemandFollowingControl
+    demand_component: nh3_load_demand

h2integrate/core/test/test_slc_framework.py

@@ -5,6 +5,9 @@
 from pytest import fixture
 
 from h2integrate import H2IntegrateModel, load_yaml
+from h2integrate.control.control_strategies.system_level.system_level_control_base import (
+    SystemLevelControlBase,
+)
 
 
 @fixture
@@ -301,3 +304,52 @@ def test_slc_topology_nh3_system_with_h2_demand(plant_config, tech_control_class
 
     with subtests.test("All upstream techs in SLC graph"):
         assert upstream_techs == slc_techs.difference({demand_tech})
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("connection_case", ["fake_complex_system"])
+def test_slc_topology_fake_complex_system(plant_config, tech_control_classifiers, subtests):
+    # multiple commodities are connected between electrolyzer and haber bosch
+    model = object.__new__(H2IntegrateModel)
+    model.slc = True
+    model.plant_config = plant_config
+    model.tech_control_classifiers = tech_control_classifiers
+    demand_tech = plant_config["system_level_control"]["demand_component"]
+    storage_techs = [k for k, v in tech_control_classifiers.items() if v == "storage"]
+    tech_config = make_mock_tech_config(demand_tech, storage_techs)
+    model.technology_config = tech_config
+    model.technology_graph = model.create_technology_graph(
+        plant_config.get("technology_interconnections", {})
+    )
+    slc_topology = model._classify_slc_technologies()
+
+    with subtests.test("Multi-commodity edge"):
+        pem_to_hb = model.technology_graph.edges["electrolyzer", "haber_bosch"].get("commodity")
+        assert len(pem_to_hb) == 3
+        assert set(pem_to_hb) == {"hydrogen", "oxygen", "heat"}
+
+    electrolyzer_commodities = [("electrolyzer", cmod) for cmod in ["hydrogen", "oxygen", "heat"]]
+    with subtests.test("Multi-commodity tech_to_commodity"):
+        assert all(
+            tech_cmod in slc_topology["tech_to_commodity"] for tech_cmod in electrolyzer_commodities
+        )
+
+    slc_base = object.__new__(SystemLevelControlBase)
+    input_tech_classifiers = ["fixed", "flexible", "dispatchable", "storage"]
+    input_techs = [
+        k
+        for k, v in slc_topology["tech_control_classifiers"].items()
+        if v in input_tech_classifiers
+    ]
+    feedstock_techs = [
+        k for k, v in slc_topology["tech_control_classifiers"].items() if v == "feedstock"
+    ]
+    slc_base.technology_graph = slc_topology["technology_graph"]
+    slc_base.input_techs = set(input_techs)
+    slc_base.feedstock_comps = feedstock_techs
+
+    upstream_techs = slc_base.get_upstream_techs_for_commodity(
+        "haber_bosch", "hydrogen", include_feedstock_sources=True
+    )
+    with subtests.test("Electrolyzer is upstream of Haber Bosch"):
+        assert upstream_techs == ["electrolyzer"]