From ed2afd967fa57fb0758b9e821e10461586d0ade1 Mon Sep 17 00:00:00 2001 From: avdudchenko <33663878+avdudchenko@users.noreply.github.com> Date: Thu, 23 Apr 2026 14:36:32 -0400 Subject: [PATCH 1/3] add arc connecto and tests --- pyomo/network/port.py | 47 ++++++++++++++++++++++- pyomo/network/tests/test_arc.py | 66 ++++++++++++++++++++++++++++++++ pyomo/network/tests/test_port.py | 46 ++++++++++++++++++++++ 3 files changed, 158 insertions(+), 1 deletion(-) diff --git a/pyomo/network/port.py b/pyomo/network/port.py index f56842096d5..1c0d96d9fd2 100644 --- a/pyomo/network/port.py +++ b/pyomo/network/port.py @@ -31,6 +31,7 @@ from pyomo.network.util import create_var, tighten_var_domain + logger = logging.getLogger('pyomo.network') @@ -282,7 +283,6 @@ def get_split_fraction(self, arc): else: return res - class _PortData(metaclass=RenamedClass): __renamed__new_class__ = PortData __renamed__version__ = '6.7.2' @@ -335,6 +335,7 @@ def __init__(self, *args, **kwd): self._initialize = kwd.pop('initialize', {}) self._implicit = kwd.pop('implicit', {}) self._extends = kwd.pop('extends', None) + self._auto_created_arcs = [] kwd.setdefault('ctype', Port) IndexedComponent.__init__(self, *args, **kwd) @@ -375,6 +376,50 @@ def construct(self, data=None): timer.report() + def connect_to(self,port, block=None): + """Method for connecting current port to another port via an Arc + Args: + port: Port + The destination port to connect to + block: Block, optional + The block on which to construct the Arc. If None, the Arc will be constructed on the port parent block + """ + #NOTE import Arc here to avoid circular import issues, this is the only place in the Port class where Arc is needed + from pyomo.network.arc import Arc + + if block is None: + block=self.parent_block() + if block is None: + raise ValueError( + "Cannot connect Port '%s' to Port '%s' because neither port has a parent block. Please specify a block to construct the Arc on." + % (self.name, port.name) + ) + def get_safe_name(name): + return ( + name.replace(".", "_") + .replace("-", "") + .replace("[", "_") + .replace("]", "") + ) + + current_port_name=get_safe_name(self.name) + dest_port_name=get_safe_name(port.name) + + arc_name=f"{current_port_name}_to_{dest_port_name}" + block.add_component(arc_name, Arc(source=self, destination=port)) + created_arc=block.find_component(arc_name) + if created_arc is None: + raise RuntimeError( + f"Failed to create Arc '{arc_name}' connecting Port '{self.name}' to Port '{port.name}'." + ) + logger.info(f"Created Arc '{arc_name}' connecting Port '{self.name}' to Port '{port.name}'.") + self._auto_created_arcs.append(created_arc) + + def get_created_arcs(self): + """Returns a list of Arcs that were automatically created by the `connect_to` method, + this list can be used for simple propagation or management of automatically created arcs""" + return self._auto_created_arcs + def _initialize_members(self, initSet): for idx in initSet: tmp = self[idx] diff --git a/pyomo/network/tests/test_arc.py b/pyomo/network/tests/test_arc.py index f3c26285dc7..f03baf32b22 100644 --- a/pyomo/network/tests/test_arc.py +++ b/pyomo/network/tests/test_arc.py @@ -1986,6 +1986,72 @@ def test_clone(self): m2.y.value = 1.25 self.assertAlmostEqual(value(c.body), 0) + def test_clone(self): + m = ConcreteModel() + m.x = Var() + m.y = Var() + m.p1 = Port() + m.p2 = Port() + m.p1.add(m.x, 'v') + m.p2.add(m.y, 'v') + m.arc = Arc(source=m.p1, destination=m.p2) + + m2 = m.clone() + self.assertEqual(len(m2.p1.arcs()), 1) + self.assertEqual(len(m2.p2.arcs()), 1) + self.assertIs(m2.p1.arcs()[0], m2.arc) + self.assertIs(m2.p2.arcs()[0], m2.arc) + + self.assertIsNot(m2.p1.arcs()[0], m.arc) + self.assertIsNot(m2.p2.arcs()[0], m.arc) + + TransformationFactory('network.expand_arcs').apply_to(m2) + all_cons = list(m2.component_data_objects(Constraint)) + self.assertEqual(len(all_cons), 1) + c = all_cons[0] + self.assertAlmostEqual(value(c.lower), 0) + self.assertAlmostEqual(value(c.upper), 0) + c_vars = ComponentSet(identify_variables(c.body)) + self.assertIn(m2.x, c_vars) + self.assertIn(m2.y, c_vars) + self.assertNotIn(m.x, c_vars) + self.assertNotIn(m.y, c_vars) + m2.x.value = 1.25 + m2.y.value = 1.25 + self.assertAlmostEqual(value(c.body), 0) + + def test_expand_auto_connect(self): + m = ConcreteModel() + m.x = Var() + m.prt = Port() + m.prt.add(m.x, "a") + m.prt.connect_to(m.prt) + created_arc=m.find_component("prt_to_prt") + self.assertEqual(len(list(m.component_objects(Constraint))), 0) + self.assertEqual(len(list(m.component_data_objects(Constraint))), 0) + + TransformationFactory('network.expand_arcs').apply_to(m) + + self.assertEqual(len(list(m.component_objects(Constraint))), 1) + self.assertEqual(len(list(m.component_data_objects(Constraint))), 1) + self.assertFalse(created_arc.active) + blk = m.component('prt_to_prt_expanded') + self.assertTrue(blk.active) + self.assertTrue(blk.component('a_equality').active) + + os = StringIO() + blk.pprint(ostream=os) + self.assertEqual( + os.getvalue(), + """prt_to_prt_expanded : Size=1, Index=None, Active=True + 1 Constraint Declarations + a_equality : Size=1, Index=None, Active=True + Key : Lower : Body : Upper : Active + None : 0.0 : x - x : 0.0 : True + + 1 Declarations: a_equality +""", + ) if __name__ == "__main__": unittest.main() diff --git a/pyomo/network/tests/test_port.py b/pyomo/network/tests/test_port.py index f62a2b8edd8..b5b175f7325 100644 --- a/pyomo/network/tests/test_port.py +++ b/pyomo/network/tests/test_port.py @@ -16,6 +16,7 @@ from pyomo.environ import ( ConcreteModel, AbstractModel, + Block, Var, Set, NonNegativeReals, @@ -553,5 +554,50 @@ def _IN(m, i): ) + def test_auto_connect(self): + m = ConcreteModel() + m.x = Var() + m.y = Var() + m.p1 = Port() + m.p1.add(m.x) + m.p2 = Port() + m.p2.add(m.y) + m.p1.connect_to(m.p2) + + self.assertIs(m.p1.x, m.x) + self.assertIs(m.p2.y, m.y) + assert m.find_component('p1_to_p2') is not None + + def test_auto_connect_with_block(self): + m = ConcreteModel() + m.block_a=Block() + m.block_b=Block() + m.block_a.x = Var() + m.block_b.y = Var() + m.block_a.p1 = Port() + m.block_a.p1.add(m.block_a.x) + m.block_b.p2 = Port() + m.block_b.p2.add(m.block_b.y) + m.block_a.p1.connect_to(m.block_b.p2) + + self.assertIs(m.block_a.p1.x, m.block_a.x) + self.assertIs(m.block_b.p2.y, m.block_b.y) + assert m.block_a.find_component('block_a_p1_to_block_b_p2') is not None + + def test_auto_connect_with_indexed_block(self): + m = ConcreteModel() + m.block=Block([1,2]) + m.block[1].x = Var() + m.block[2].y = Var() + m.block[1].p1 = Port() + m.block[1].p1.add(m.block[1].x) + m.block[2].p2 = Port() + m.block[2].p2.add(m.block[2].y) + m.block[1].p1.connect_to(m.block[2].p2) + + self.assertIs(m.block[1].p1.x, m.block[1].x) + self.assertIs(m.block[2].p2.y, m.block[2].y) + assert m.block[1].find_component('block_1_p1_to_block_2_p2') is not None + if __name__ == "__main__": unittest.main() From af5fa3f1ee49c6134b04690ca757a3f13446fc33 Mon Sep 17 00:00:00 2001 From: avdudchenko <33663878+avdudchenko@users.noreply.github.com> Date: Thu, 23 Apr 2026 14:55:27 -0400 Subject: [PATCH 2/3] add tsts for get arc --- pyomo/network/port.py | 13 ++++++++++--- pyomo/network/tests/test_port.py | 9 +++++++-- 2 files changed, 17 insertions(+), 5 deletions(-) diff --git a/pyomo/network/port.py b/pyomo/network/port.py index 1c0d96d9fd2..22d7576b5dd 100644 --- a/pyomo/network/port.py +++ b/pyomo/network/port.py @@ -416,9 +416,16 @@ def get_safe_name(name): self._auto_created_arcs.append(created_arc) def get_created_arcs(self): - """Returns a list of Arcs that were automatically created by the `connect_to` method, - this list can be used for simple propagation or management of automatically created arcs""" - return self._auto_created_arcs + """Returns a singel or list of Arcs that were automatically created by the `connect_to` method, + this list can be used for simple propagation or management of automatically created arcs, + if no Arc is created None is returned.""" + if len(self._auto_created_arcs) == 0: + logger.warning(f"No automatically created arcs found for Port '{self.name}'.") + return None + elif len(self._auto_created_arcs) == 1: + return self._auto_created_arcs[0] + else: + return self._auto_created_arcs def _initialize_members(self, initSet): for idx in initSet: diff --git a/pyomo/network/tests/test_port.py b/pyomo/network/tests/test_port.py index b5b175f7325..0c3d1807332 100644 --- a/pyomo/network/tests/test_port.py +++ b/pyomo/network/tests/test_port.py @@ -567,7 +567,8 @@ def test_auto_connect(self): self.assertIs(m.p1.x, m.x) self.assertIs(m.p2.y, m.y) assert m.find_component('p1_to_p2') is not None - + assert m.p1.get_created_arcs() is m.find_component('p1_to_p2') + assert m.p2.get_created_arcs() is None def test_auto_connect_with_block(self): m = ConcreteModel() m.block_a=Block() @@ -583,7 +584,8 @@ def test_auto_connect_with_block(self): self.assertIs(m.block_a.p1.x, m.block_a.x) self.assertIs(m.block_b.p2.y, m.block_b.y) assert m.block_a.find_component('block_a_p1_to_block_b_p2') is not None - + assert m.block_a.p1.get_created_arcs() is m.block_a.find_component('block_a_p1_to_block_b_p2') + assert m.block_b.p2.get_created_arcs() is None def test_auto_connect_with_indexed_block(self): m = ConcreteModel() m.block=Block([1,2]) @@ -598,6 +600,9 @@ def test_auto_connect_with_indexed_block(self): self.assertIs(m.block[1].p1.x, m.block[1].x) self.assertIs(m.block[2].p2.y, m.block[2].y) assert m.block[1].find_component('block_1_p1_to_block_2_p2') is not None + + assert m.block[1].p1.get_created_arcs() is m.block[1].find_component('block_1_p1_to_block_2_p2') + assert m.block[2].p2.get_created_arcs() is None if __name__ == "__main__": unittest.main() From 177a6045d21e12c5847d0aa72a926328a59a2101 Mon Sep 17 00:00:00 2001 From: avdudchenko <33663878+avdudchenko@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:51:58 -0400 Subject: [PATCH 3/3] black formating --- pyomo/network/port.py | 40 ++++++++++++++++++-------------- pyomo/network/tests/test_arc.py | 3 ++- pyomo/network/tests/test_port.py | 34 ++++++++++++++++----------- 3 files changed, 45 insertions(+), 32 deletions(-) diff --git a/pyomo/network/port.py b/pyomo/network/port.py index 22d7576b5dd..62c5dcfc50c 100644 --- a/pyomo/network/port.py +++ b/pyomo/network/port.py @@ -283,6 +283,7 @@ def get_split_fraction(self, arc): else: return res + class _PortData(metaclass=RenamedClass): __renamed__new_class__ = PortData __renamed__version__ = '6.7.2' @@ -376,24 +377,25 @@ def construct(self, data=None): timer.report() - def connect_to(self,port, block=None): + def connect_to(self, port, block=None): """Method for connecting current port to another port via an Arc Args: port: Port The destination port to connect to - block: Block, optional + block: Block, optional The block on which to construct the Arc. If None, the Arc will be constructed on the port parent block """ - #NOTE import Arc here to avoid circular import issues, this is the only place in the Port class where Arc is needed + # NOTE import Arc here to avoid circular import issues, this is the only place in the Port class where Arc is needed from pyomo.network.arc import Arc if block is None: - block=self.parent_block() + block = self.parent_block() if block is None: raise ValueError( "Cannot connect Port '%s' to Port '%s' because neither port has a parent block. Please specify a block to construct the Arc on." % (self.name, port.name) ) + def get_safe_name(name): return ( name.replace(".", "_") @@ -401,32 +403,36 @@ def get_safe_name(name): .replace("[", "_") .replace("]", "") ) - - current_port_name=get_safe_name(self.name) - dest_port_name=get_safe_name(port.name) - - arc_name=f"{current_port_name}_to_{dest_port_name}" + + current_port_name = get_safe_name(self.name) + dest_port_name = get_safe_name(port.name) + + arc_name = f"{current_port_name}_to_{dest_port_name}" block.add_component(arc_name, Arc(source=self, destination=port)) - created_arc=block.find_component(arc_name) + created_arc = block.find_component(arc_name) if created_arc is None: raise RuntimeError( f"Failed to create Arc '{arc_name}' connecting Port '{self.name}' to Port '{port.name}'." - ) - logger.info(f"Created Arc '{arc_name}' connecting Port '{self.name}' to Port '{port.name}'.") + ) + logger.info( + f"Created Arc '{arc_name}' connecting Port '{self.name}' to Port '{port.name}'." + ) self._auto_created_arcs.append(created_arc) - def get_created_arcs(self): - """Returns a singel or list of Arcs that were automatically created by the `connect_to` method, - this list can be used for simple propagation or management of automatically created arcs, + def get_connections(self): + """Returns a single or list of Arcs that were automatically created by the `connect_to` method, + this list can be used for simple propagation or management of automatically created arcs, if no Arc is created None is returned.""" if len(self._auto_created_arcs) == 0: - logger.warning(f"No automatically created arcs found for Port '{self.name}'.") + logger.warning( + f"No automatically created arcs found for Port '{self.name}'." + ) return None elif len(self._auto_created_arcs) == 1: return self._auto_created_arcs[0] else: return self._auto_created_arcs - + def _initialize_members(self, initSet): for idx in initSet: tmp = self[idx] diff --git a/pyomo/network/tests/test_arc.py b/pyomo/network/tests/test_arc.py index f03baf32b22..446b3c2db72 100644 --- a/pyomo/network/tests/test_arc.py +++ b/pyomo/network/tests/test_arc.py @@ -2026,7 +2026,7 @@ def test_expand_auto_connect(self): m.prt = Port() m.prt.add(m.x, "a") m.prt.connect_to(m.prt) - created_arc=m.find_component("prt_to_prt") + created_arc = m.find_component("prt_to_prt") self.assertEqual(len(list(m.component_objects(Constraint))), 0) self.assertEqual(len(list(m.component_data_objects(Constraint))), 0) @@ -2053,5 +2053,6 @@ def test_expand_auto_connect(self): """, ) + if __name__ == "__main__": unittest.main() diff --git a/pyomo/network/tests/test_port.py b/pyomo/network/tests/test_port.py index 0c3d1807332..1f09b1525e0 100644 --- a/pyomo/network/tests/test_port.py +++ b/pyomo/network/tests/test_port.py @@ -553,7 +553,6 @@ def _IN(m, i): """, ) - def test_auto_connect(self): m = ConcreteModel() m.x = Var() @@ -563,16 +562,17 @@ def test_auto_connect(self): m.p2 = Port() m.p2.add(m.y) m.p1.connect_to(m.p2) - + self.assertIs(m.p1.x, m.x) self.assertIs(m.p2.y, m.y) assert m.find_component('p1_to_p2') is not None - assert m.p1.get_created_arcs() is m.find_component('p1_to_p2') - assert m.p2.get_created_arcs() is None + assert m.p1.get_connections() is m.find_component('p1_to_p2') + assert m.p2.get_connections() is None + def test_auto_connect_with_block(self): m = ConcreteModel() - m.block_a=Block() - m.block_b=Block() + m.block_a = Block() + m.block_b = Block() m.block_a.x = Var() m.block_b.y = Var() m.block_a.p1 = Port() @@ -580,15 +580,18 @@ def test_auto_connect_with_block(self): m.block_b.p2 = Port() m.block_b.p2.add(m.block_b.y) m.block_a.p1.connect_to(m.block_b.p2) - + self.assertIs(m.block_a.p1.x, m.block_a.x) self.assertIs(m.block_b.p2.y, m.block_b.y) assert m.block_a.find_component('block_a_p1_to_block_b_p2') is not None - assert m.block_a.p1.get_created_arcs() is m.block_a.find_component('block_a_p1_to_block_b_p2') - assert m.block_b.p2.get_created_arcs() is None + assert m.block_a.p1.get_connections() is m.block_a.find_component( + 'block_a_p1_to_block_b_p2' + ) + assert m.block_b.p2.get_connections() is None + def test_auto_connect_with_indexed_block(self): m = ConcreteModel() - m.block=Block([1,2]) + m.block = Block([1, 2]) m.block[1].x = Var() m.block[2].y = Var() m.block[1].p1 = Port() @@ -596,13 +599,16 @@ def test_auto_connect_with_indexed_block(self): m.block[2].p2 = Port() m.block[2].p2.add(m.block[2].y) m.block[1].p1.connect_to(m.block[2].p2) - + self.assertIs(m.block[1].p1.x, m.block[1].x) self.assertIs(m.block[2].p2.y, m.block[2].y) assert m.block[1].find_component('block_1_p1_to_block_2_p2') is not None - - assert m.block[1].p1.get_created_arcs() is m.block[1].find_component('block_1_p1_to_block_2_p2') - assert m.block[2].p2.get_created_arcs() is None + + assert m.block[1].p1.get_connections() is m.block[1].find_component( + 'block_1_p1_to_block_2_p2' + ) + assert m.block[2].p2.get_connections() is None + if __name__ == "__main__": unittest.main()