Optimization backend (corneto.backend)#

Optimization backends and the runtime-selected defaults.

class corneto.backend.Backend(default_solver=None, sparse_class=<class 'scipy.sparse._csr.csr_array'>)#

Bases: ABC

Parameters:
is_available()#
Return type:

bool

version()#
Return type:

str

static get_symbols(expressions)#
Parameters:

expressions (Iterable[CExpression])

Return type:

Set[CSymbol]

abstractmethod available_solvers()#
Return type:

List[str]

abstractmethod Constant(value, name=None)#
Parameters:

  • value (Any)

  • name (str | None)

Return type:

CSymbol

abstractmethod Variable(name=None, shape=None, lb=None, ub=None, vartype=VarType.CONTINUOUS)#

Create a variable for optimization.

This method defines an optimization variable with optional bounds, type, and additional graph-related properties.

Parameters:

  • name (Optional[str]) – The name of the variable. Defaults to None.

  • shape (Optional[Tuple[int, ...]]) – The shape of the variable as a tuple. Defaults to None.

  • lb (Optional[Union[float, np.ndarray]]) – The lower bound of the variable. Can be a scalar or an array. Defaults to None.

  • ub (Optional[Union[float, np.ndarray]]) – The upper bound of the variable. Can be a scalar or an array. Defaults to None.

  • vartype (VarType) – The type of the variable (e.g., continuous, integer). Defaults to VarType.CONTINUOUS.

Returns:

The created variable symbol, to be used in further expressions or constraints.

Return type:

CSymbol

abstractmethod Parameter(name=None, shape=None, value=None)#
Parameters:
Return type:

CSymbol

TotalVariation(X, *, pairs, axis=-1, weights=None, name=None, exact=False)#

Build an entrywise graph total-variation expression.

X must be a nonempty, real affine vector or matrix expression created by this backend. Continuous, integer, and binary expressions are supported, whether or not their variables have bounds. The selected axis of X is connected by the directed pairs. Each pair contributes the destination-minus-source signed difference, summed over the other axis; its magnitude contributes to variation. Reversing a pair reverses the signed difference but leaves its magnitude unchanged.

The returned problem contains a continuous absolute-value bound and registered expressions for the signed differences, absolute bounds, per-pair variation, and weighted total variation. Add its total variation expression to an objective or constrain it explicitly; no objective is added automatically. The total scales linearly with the magnitude of X and the number of rows summed for each pair. With exact=True, binary inputs use exact XOR magnitudes, including when maximizing variation or when a pair has zero weight.

Parameters:

  • X (CExpression) – A nonempty real affine vector or matrix expression from this backend.

  • pairs (Iterable[Tuple[int, int]]) – Nonempty iterable of (source, destination) integer index pairs along axis. Self-pairs are rejected. Pair order and duplicates are preserved.

  • axis (int) – Axis of X connected by pairs; defaults to -1. Negative axes follow NumPy conventions.

  • weights (Sequence[float] | ndarray | None) – Optional one-dimensional sequence or array with one finite, nonnegative real weight per pair. Defaults to unit weights. Scalars are rejected.

  • name (str | None) – Optional stable expression-name prefix. If omitted, a unique prefix is generated.

  • exact (bool) – If true, require binary inputs and enforce exact absolute differences independently of the objective. Binary variables are used directly; affine expressions are linked to auxiliary binary selectors, constraining every entry of X to zero or one. Nonbinary symbols are rejected. Defaults to false, which retains the continuous absolute-value bound formulation.

Returns:

A problem with the linear absolute-value formulation and registered expressions named <prefix>_difference and <prefix>_abs_bound with shape (R, E), <prefix>_variation_by_pair with shape (E,), and scalar <prefix>_total_variation. Here E is the number of pairs and R is the size of the other axis (or one for a vector).

Return type:

ProblemDef

Problem(constraints=None, objectives=None, expressions=None, weights=None, direction=Direction.MIN)#
Parameters:

  • constraints (CExpression | List[CExpression] | None)

  • objectives (CExpression | List[CExpression] | None)

  • expressions (Dict[str, CExpression] | None)

  • weights (float | List[float] | None)

  • direction (Direction)

Return type:

ProblemDef

abstractmethod build(p)#
Parameters:

p (ProblemDef)

Return type:

Any

solve(p, solver=None, max_seconds=None, warm_start=False, verbosity=0, **options)#
Parameters:

  • p (ProblemDef)

  • solver (str | Solver | None)

  • max_seconds (int | None)

  • warm_start (bool)

  • verbosity (int)

Flow(g, lb=0, ub=10, n_flows=1, values=False, shared_bounds=False, varname='_flow', create_nonzero_indicators=False, alias_flow='flow', alias_flow_ipos='positive_flow', alias_flow_ineg='negative_flow', alias_nonzero_flow='with_flow', indicator_tolerance=0.0001, force_matrix=False)#
Parameters:
Return type:

ProblemDef

Acyclic0(g, P, indicator_positive_var_name=None, indicator_negative_var_name=None, acyclic_var_name='_dag_layer', max_parents=None, vertex_lb_dist=None, vertex_ub_dist=None)#

Create Acyclicity Constraint.

This function creates acyclicity constraints, ensuring that the selected edges form an acyclic graph, meaning there are no cycles on the given property. Acyclicity can be applied, for example, over flow constraints or signal properties.

Parameters:

  • g (BaseGraph) – The graph that defines the problem.

  • P (ProblemDef) – The problem definition.

  • indicator_positive_var_name (str) – The name of the indicator variable, i.e., which edges are selected. Default is EXPR_NAME_FLOW_IPOS.

  • indicator_negative_var_name (str, optional) – The name of the indicator variable for negative flows. Default is None. If a negative flow appears, the source and target nodes of the edge are reversed. For example, A->B with positive flow implies order(B) > order(A), with negative flow it implies order(A) > order(B).

  • acyclic_var_name (str, optional) – The name of the acyclic variable. Default is VAR_DAG.

  • max_parents (Optional[Union[int, Dict[Any, int]]], optional) – The maximum number of parents per node. If an integer is provided, the maximum number of parents is the same for all nodes. If a dictionary is provided, the maximum number of parents can be different for each node. Default is None.

  • Returns

  • -------

  • ProblemDef – The problem definition with acyclic constraints.

  • Raises

  • ------

  • NotImplementedError – If hyperedges are used.

  • vertex_lb_dist (List[Dict[Any, int]] | None)

  • vertex_ub_dist (List[Dict[Any, int]] | None)

Return type:

ProblemDef

Acyclic(g, P, indicator_positive_var_name=None, indicator_negative_var_name=None, acyclic_var_name='_dag_layer', max_parents=None, vertex_lb_dist=None, vertex_ub_dist=None)#

Create Acyclicity Constraint.

This function creates acyclicity constraints, ensuring that the selected edges form an acyclic graph, meaning there are no cycles on the given property. Acyclicity can be applied, for example, over flow constraints or signal properties.

Parameters:

  • g (BaseGraph) – The graph that defines the problem.

  • P (ProblemDef) – The problem definition.

  • indicator_positive_var_name (str) – The name of the indicator variable, i.e., which edges are selected. Default is EXPR_NAME_FLOW_IPOS.

  • indicator_negative_var_name (str, optional) – The name of the indicator variable for negative flows. Default is None. If a negative flow appears, the source and target nodes of the edge are reversed. For example, A->B with positive flow implies order(B) > order(A), with negative flow it implies order(A) > order(B).

  • acyclic_var_name (str, optional) – The name of the acyclic variable. Default is VAR_DAG.

  • max_parents (Optional[Union[int, Dict[Any, int]]], optional) – The maximum number of parents per node. If an integer is provided, the maximum number of parents is the same for all nodes. If a dictionary is provided, the maximum number of parents can be different for each node. Default is None.

  • vertex_lb_dist (Optional[List[Dict[Any, int]]], optional) – A list (one entry per experiment) of dictionaries that assign a lower bound (minimum layer/distance) for each vertex.

  • vertex_ub_dist (Optional[List[Dict[Any, int]]], optional) – A list (one entry per experiment) of dictionaries that assign an upper bound (maximum layer/distance) for each vertex.

  • Returns

  • -------

  • ProblemDef – The problem definition with acyclic constraints.

  • Raises

  • ------

  • NotImplementedError – If hyperedges are used.

Return type:

ProblemDef

AcyclicFlow(g, lb=0, ub=10, values=False, max_parents=None, vertex_lb_dist=None, varname='_flow', alias_flow='flow', alias_flow_ipos='positive_flow', alias_flow_ineg='negative_flow', alias_nonzero_flow='with_flow', indicator_tolerance=0.0001)#
Parameters:
Return type:

ProblemDef

Indicator(V, indexes=None, suffix='_i', name=None)#
Parameters:
Return type:

ProblemDef

ThresholdIndicator(V, *, indexes=None, epsilon=1.0, name=None, positive_name=None, negative_name=None)#

Create optional positive and negative threshold implications.

A positive binary equal to one requires the bounded value to be at least epsilon. A negative binary equal to one requires it to be at most -epsilon. When both binaries are zero, the original bounds remain available, including values between the thresholds. Reaching a threshold does not require its binary to be one.

epsilon must be finite and positive, and may be an array broadcastable to the selected value shape. A vector of thresholds also accepts a backend’s single-column representation of a vector. The returned problem registers the sum of the mutually exclusive binaries under name (default <variable>_threshold), with directional symbols named <name>_positive and <name>_negative unless explicitly supplied. Implications use linear constraints derived from the finite bounds.

Parameters:
Return type:

ProblemDef

ExactSupport(V, *, selected=None, indexes=None, epsilon=1.0, nonnegative=False, name=None, positive_name=None, negative_name=None)#

Link a bounded value to binary structural support with a gap.

selected == 0 forces the value to zero. selected == 1 forces its magnitude to be at least epsilon. For nonnegative values this uses one binary per entry. Signed values use mutually exclusive positive and negative binaries and expose their sum as selected.

epsilon may be a scalar or an array explicitly broadcastable to the selected value shape. “Exact” refers to this mathematical minimum-magnitude gap; it is not a solver-tolerance guarantee, and values smaller than the requested gap may still be returned by a numerically inaccurate solve and should be checked by the caller.

An existing binary selector symbol can be supplied to avoid introducing a redundant binary variable, which is useful when a method already has a shared structural edge-selection variable. An arbitrary selector expression is linked to an auxiliary binary selector, so fractional expression values make the model infeasible instead of relaxing the support constraints.

Parameters:
Return type:

ProblemDef

SelectedFlow(g, *, lb=0, ub=10, n_flows=1, edge_indices=None, flow_blocks=None, selector_groups=None, epsilon=1.0, exact_support=True, selected=None, acyclic_graph=None, max_parents=None, flow_name='flow', selected_by_flow_name='selected_by_flow', selected_by_group_name=None, selected_any_name='selected_any', dag_name='_dag_layer')#

Create bounded flows with exact per-flow and shared support.

Flow conservation is imposed on g. Exact support binaries are created only for edge_indices so boundary edges do not consume unnecessary integer variables. Rows whose selected flow bounds are nonnegative use one binary per flow; rows that permit negative flow automatically use mutually exclusive positive and negative binaries. Set exact_support=False to use one bounded indicator per entry: nonzero flow still implies selection, but a selected entry may carry zero flow. This smaller formulation is suitable when a minimizing objective makes such false selections unattractive and flow direction is not otherwise required.

flow_blocks can map multiple rectangular regions of the flow matrix to the same logical edge ordering. selector_groups can then map block columns to shared selector columns. This supports layouts such as forward and reversed edge blocks without indicators on the unused off-diagonal blocks. Set selected_any_name=None when a union across selector columns is not needed.

The union across flows can be linked to an existing shared selector. With signed acyclic flows, direction-specific unions are used so a negative flow orders the edge in reverse. No redundant structural-union binary is created in that case.

Parameters:
Return type:

ProblemDef

NonZeroIndicator(V, *args, indexes=None, suffix_pos='_ipos', suffix_neg='_ineg', tolerance=0.001)#
Parameters:
Return type:

ProblemDef

linear_or(x, axis=None, varname='or', ignore_type=False)#
Parameters:

  • x (CExpression)

  • axis (int | None)

Return type:

ProblemDef

linear_and(x, axis=None, varname='and')#
Parameters:

  • x (CExpression)

  • axis (int | None)

Return type:

ProblemDef

linear_xor(x, axis=None, varname='xor', ignore_type=False)#
Parameters:

  • x (CExpression)

  • axis (int | None)

Return type:

ProblemDef

vstack(arg_list)#
Parameters:

arg_list (Iterable[CExpression])

Return type:

CExpression

hstack(arg_list)#
Parameters:

arg_list (Iterable[CExpression])

Return type:

CExpression

zero_function()#
Return type:

CExpression

class corneto.backend.CvxpyBackend(default_solver=None, sparse_class=<class 'scipy.sparse._csr.csr_array'>)#

Bases: Backend

Parameters:
available_solvers()#
Return type:

List[str]

Constant(value, name=None)#
Parameters:

  • value (Any)

  • name (str | None)

Return type:

CSymbol

Variable(name=None, shape=None, lb=None, ub=None, vartype=VarType.CONTINUOUS)#

Create a variable for optimization.

This method defines an optimization variable with optional bounds, type, and additional graph-related properties.

Parameters:

  • name (Optional[str]) – The name of the variable. Defaults to None.

  • shape (Optional[Tuple[int, ...]]) – The shape of the variable as a tuple. Defaults to None.

  • lb (Optional[Union[float, np.ndarray]]) – The lower bound of the variable. Can be a scalar or an array. Defaults to None.

  • ub (Optional[Union[float, np.ndarray]]) – The upper bound of the variable. Can be a scalar or an array. Defaults to None.

  • vartype (VarType) – The type of the variable (e.g., continuous, integer). Defaults to VarType.CONTINUOUS.

Returns:

The created variable symbol, to be used in further expressions or constraints.

Return type:

CSymbol

Parameter(name=None, shape=None, value=None)#
Parameters:
Return type:

CSymbol

build(p)#
Parameters:

p (ProblemDef)

Return type:

Any

class corneto.backend.PicosBackend(default_solver=None)#

Bases: Backend

Parameters:

default_solver (str | None)

available_solvers()#
Return type:

List[str]

build(p)#
Parameters:

p (ProblemDef)

Return type:

Any

Constant(value, name=None)#
Parameters:

  • value (Any)

  • name (str | None)

Return type:

CSymbol

Variable(name=None, shape=None, lb=None, ub=None, vartype=VarType.CONTINUOUS, variable=True)#

Create a variable for optimization.

This method defines an optimization variable with optional bounds, type, and additional graph-related properties.

Parameters:

  • name (Optional[str]) – The name of the variable. Defaults to None.

  • shape (Optional[Tuple[int, ...]]) – The shape of the variable as a tuple. Defaults to None.

  • lb (Optional[Union[float, np.ndarray]]) – The lower bound of the variable. Can be a scalar or an array. Defaults to None.

  • ub (Optional[Union[float, np.ndarray]]) – The upper bound of the variable. Can be a scalar or an array. Defaults to None.

  • vartype (VarType) – The type of the variable (e.g., continuous, integer). Defaults to VarType.CONTINUOUS.

  • variable (bool)

Returns:

The created variable symbol, to be used in further expressions or constraints.

Return type:

CSymbol

Parameter(name=None, shape=None, value=None)#
Parameters:
Return type:

CSymbol

class corneto.backend.VarType(value)#

Bases: str, Enum

INTEGER = 'integer'#
CONTINUOUS = 'continuous'#
BINARY = 'binary'#
corneto.backend.available_backends()#

Return the installed and usable optimization backends.

Total variation#

Backend.TotalVariation builds entrywise L1 graph variation for a nonempty, real affine CORNETO vector or matrix expression created by the same backend. Continuous, integer, and binary expressions are supported, with or without variable bounds. For a vector, the connected axis is its only axis. For a matrix, axis selects the connected dimension and defaults to -1; negative axes follow NumPy conventions. The remaining dimension is summed.

pairs must contain at least one directed (source, destination) pair with integer indices in range. Empty input and self-pairs are rejected. Duplicate and reversed pairs each contribute independently. The registered signed difference is destination minus source; reversing a pair reverses that signed value while leaving its magnitude unchanged. Each pair’s variation is the sum of these magnitudes across the other axis. The total variation applies the pair weights to those sums. The optional one-dimensional weights sequence or array must contain one finite, nonnegative real value per pair; scalars are rejected and omitted weights are all one.

Every result registers <name>_difference and <name>_abs_bound with shape (R, E), <name>_variation_by_pair with shape (E,), and scalar <name>_total_variation. Here E is the number of pairs and R is the size of the other axis, or one for a vector. When name is omitted, a unique prefix is generated. Explicit prefixes should be distinct in a composed problem. No objective is added automatically.

By default (exact=False), the primitive adds continuous variables D and constraints D >= difference and D >= -difference with D >= 0. Thus the registered total variation is an upper bound until an objective or another constraint tightens D; X does not need variable bounds. A positive coefficient in a minimization objective makes positively weighted entries tight at an optimum when other uses permit it; zero weights need not be tight. A budget constraint total_variation <= budget still gives the exact projected upper bound on actual variation. Maximizing total variation or lower-bounding it does not force D to equal the absolute difference, so diagnostics should be recomputed from the solved difference expression. In a maximization objective, subtract a positive variation penalty. Entrywise L1 graph variation encourages connected entries to be piecewise constant. Its scale depends on the magnitude and units of X, the number of entries summed across the other axis, and the chosen weights; multiplying X by a scalar multiplies the variation by its absolute value.

For binary selections, exact=True makes every absolute difference equal to the pairwise XOR, independently of the objective and including zero-weight pairs. Binary variables are accepted directly. Affine expressions (including slices and sums of selection indicators) are linked to auxiliary binary selectors, explicitly restricting every input entry to zero or one. Nonbinary symbols are rejected; this mode does not compute exact absolute values for general continuous or integer inputs.

For binary endpoints a and b, exact mode adds D <= a + b and D <= 2 - a - b to the usual lower bounds. These force D = abs(a - b) while keeping D continuous. The registered names and shapes are unchanged. This permits maximizing switch counts or imposing a lower bound on switching. For example, after constraining the union size to its previously obtained optimum, use the following as the diagnostic’s sole objective (CORNETO minimizes by default):

tv = backend.TotalVariation(
    selected,
    pairs=[(0, 1), (1, 2)],
    axis=1,
    exact=True,
    name="switches",
)
problem += tv
problem.add_objective(-tv.expr.switches_total_variation)

For a continuous expression whose rows represent signals over three ordered conditions, connect adjacent columns and add the variation penalty to any existing model objective:

tv = backend.TotalVariation(
    X,
    pairs=[(0, 1), (1, 2)],
    axis=1,
    weights=[1.0, 2.0],
    name="conditions",
)
problem += tv
problem.add_objective(tv.expr.conditions_total_variation, weight=0.5)

The same formulation works for binary expressions over an arbitrary condition graph. To constrain the total amount of switching, merge the returned problem and add a budget instead of adding an objective:

tv = backend.TotalVariation(
    selected,
    pairs=[(0, 1), (0, 2), (2, 3)],
    axis=0,
    name="switches",
)
problem += tv
problem += tv.expr.switches_total_variation <= budget