fix(acyclic): enforce reverse-edge ordering and add small-graph regression tests

Author: pablormier

corneto/backend/_base.py

@@ -1048,8 +1048,8 @@ def Acyclic0(
             if In is not None:
                 # NOTE: Negative flows eq. to reversed directed edge
                 # Get edges s->t that can have a positive flow
-                if hasattr(In, "lb"):
-                    e_neg = [(i, g.get_edge(i)) for i in np.flatnonzero(In.lb < 0)]
+                if hasattr(In, "ub"):
+                    e_neg = [(i, g.get_edge(i)) for i in np.flatnonzero(In.ub > 0)]
                     e_ix = np.array([i for i, (s, t) in e_neg if len(s) > 0 and len(t) > 0])
                 else:
                     e_ix = np.array([i for i, (s, t) in enumerate(g.E) if len(s) > 0 and len(t) > 0])
@@ -1192,9 +1192,9 @@ def Acyclic(
 
             if In is not None:
                 # Negative flows are handled as reversed directed edges.
-                if hasattr(In, "lb"):
-                    lb = In.lb if len(In.shape) == 1 else In.lb[:, i_sample]
-                    e_neg = [(i, g.get_edge(i)) for i in np.flatnonzero(lb < 0)]
+                if hasattr(In, "ub"):
+                    ub = In.ub if len(In.shape) == 1 else In.ub[:, i_sample]
+                    e_neg = [(i, g.get_edge(i)) for i in np.flatnonzero(ub > 0)]
                     e_ix = np.array([i for i, (s, t) in e_neg if s and t])
                 else:
                     e_ix = np.array([i for i, (s, t) in enumerate(g.E) if s and t])

tests/test_acyclic_small_graphs.py

@@ -0,0 +1,90 @@
+import numpy as np
+
+from corneto._graph import EdgeType, Graph
+from corneto.backend import PicosBackend
+
+
+def _build_acyclic_problem(backend, graph, lb, ub):
+    problem = backend.Flow(graph, lb=lb, ub=ub)
+    problem += backend.NonZeroIndicator(problem.expr._flow, tolerance=1e-6)
+    problem = backend.Acyclic(
+        graph,
+        problem,
+        indicator_positive_var_name="_flow_ipos",
+        indicator_negative_var_name="_flow_ineg",
+    )
+    return problem
+
+
+def _solve_and_assert_infeasible(problem, backend):
+    if isinstance(backend, PicosBackend):
+        problem.solve(solver="glpk", primals=False)
+    else:
+        problem.solve()
+    assert problem.expr._flow.value is None
+
+
+def test_acyclic_blocks_forced_directed_2cycle(backend):
+    graph = Graph()
+    graph.add_edge("A", "B", type=EdgeType.DIRECTED)
+    graph.add_edge("B", "A", type=EdgeType.DIRECTED)
+
+    problem = _build_acyclic_problem(backend, graph, lb=0, ub=10)
+    problem += problem.expr._flow_ipos[0] == 1
+    problem += problem.expr._flow_ipos[1] == 1
+
+    _solve_and_assert_infeasible(problem, backend)
+
+
+def test_acyclic_blocks_forced_directed_plus_reversible_2cycle(backend):
+    graph = Graph()
+    graph.add_edge("A", "B", type=EdgeType.DIRECTED)
+    graph.add_edge("A", "B", type=EdgeType.UNDIRECTED)
+
+    lb = np.array([0, -10])
+    ub = np.array([10, 10])
+    problem = _build_acyclic_problem(backend, graph, lb=lb, ub=ub)
+    problem += problem.expr._flow_ipos[0] == 1
+    problem += problem.expr._flow_ineg[1] == 1
+
+    _solve_and_assert_infeasible(problem, backend)
+
+
+def test_acyclic_blocks_forced_mixed_3cycle(backend):
+    graph = Graph()
+    graph.add_edge("A", "B", type=EdgeType.DIRECTED)
+    graph.add_edge("B", "C", type=EdgeType.DIRECTED)
+    graph.add_edge("A", "C", type=EdgeType.UNDIRECTED)
+
+    lb = np.array([0, 0, -10])
+    ub = np.array([10, 10, 10])
+    problem = _build_acyclic_problem(backend, graph, lb=lb, ub=ub)
+    # Force A->B, B->C and C->A (negative use of A<->C).
+    problem += problem.expr._flow_ipos[0] == 1
+    problem += problem.expr._flow_ipos[1] == 1
+    problem += problem.expr._flow_ineg[2] == 1
+
+    _solve_and_assert_infeasible(problem, backend)
+
+
+def test_acyclic_enforces_constraints_per_sample(backend):
+    graph = Graph()
+    graph.add_edge("A", "B", type=EdgeType.DIRECTED)
+    graph.add_edge("A", "B", type=EdgeType.UNDIRECTED)
+
+    lb = np.array([0, -10])
+    ub = np.array([10, 10])
+    problem = backend.Flow(graph, lb=lb, ub=ub, n_flows=2)
+    problem += backend.NonZeroIndicator(problem.expr._flow, tolerance=1e-6)
+    problem = backend.Acyclic(
+        graph,
+        problem,
+        indicator_positive_var_name="_flow_ipos",
+        indicator_negative_var_name="_flow_ineg",
+    )
+
+    # Force A->B and B->A (negative use of reversible edge) only in sample 0.
+    problem += problem.expr._flow_ipos[0, 0] == 1
+    problem += problem.expr._flow_ineg[1, 0] == 1
+
+    _solve_and_assert_infeasible(problem, backend)