R/137_reductions_canonicalization.R

In CVXR: Disciplined Convex Optimization

#####
## DO NOT EDIT THIS FILE!! EDIT THE SOURCE INSTEAD: rsrc_tree/reductions/canonicalization.R
#####

## CVXPY SOURCE: reductions/canonicalization.py
## Canonicalization -- recursive expression tree canonicalization
##
## This is the base class for expression-level reductions. It walks the
## expression tree bottom-up and dispatches via dcp_canonicalize (S7 generic).
##
## Key methods are implemented as plain R functions (.canonicalize_tree,
## .canonicalize_expr) rather than S7 generics to avoid dispatch overhead
## during recursion.


# -- Canonicalization class ------------------------------------------
## CVXPY SOURCE: canonicalization.py lines 25-157

Canonicalization <- new_class("Canonicalization", parent = Reduction,
  package = "CVXR",
  constructor = function() {
    new_object(S7_object(),
      .cache = new.env(parent = emptyenv())
    )
  }
)

## apply: canonicalize objective + all constraints
## CVXPY SOURCE: canonicalization.py lines 55-74
method(reduction_apply, Canonicalization) <- function(x, problem, ...) {
  inverse_data <- InverseData(problem)

  ## Canonicalize objective
  obj_result <- .canonicalize_tree(problem@objective)
  canon_objective <- obj_result[[1L]]

  ## Canonicalize each constraint -- collect chunks, flatten once
  n_cons <- length(problem@constraints)
  all_chunks <- vector("list", n_cons + 1L)
  all_chunks[[1L]] <- obj_result[[2L]]
  for (i in seq_len(n_cons)) {
    con <- problem@constraints[[i]]
    con_result <- .canonicalize_tree(con)
    all_chunks[[i + 1L]] <- c(con_result[[2L]], list(con_result[[1L]]))
    ## Store constraint ID mapping
    assign(as.character(con@id), con_result[[1L]]@id,
           envir = inverse_data@cons_id_map)
  }
  canon_constraints <- unlist(all_chunks, recursive = FALSE)
  if (is.null(canon_constraints)) canon_constraints <- list()

  new_problem <- Problem(canon_objective, canon_constraints)
  list(new_problem, inverse_data)
}

## invert: map solution back through ID maps
## CVXPY SOURCE: canonicalization.py lines 76-84
method(reduction_invert, Canonicalization) <- function(x, solution, inverse_data, ...) {
  ## Remap dual variables: cons_id_map maps old_id -> new_id
  ## We need to reverse: find solution@dual_vars[[as.character(new_id)]]
  ## and store under as.character(old_id)
  if (length(solution@dual_vars) > 0L) {
    dvars <- list()
    old_ids <- ls(inverse_data@cons_id_map, all.names = TRUE)
    for (old_id in old_ids) {
      new_id <- as.character(get(old_id, envir = inverse_data@cons_id_map))
      if (!is.null(solution@dual_vars[[new_id]])) {
        dvars[[old_id]] <- solution@dual_vars[[new_id]]
      }
    }
    return(Solution(status      = solution@status,
                    opt_val     = solution@opt_val,
                    primal_vars = solution@primal_vars,
                    dual_vars   = dvars,
                    attr        = solution@attr))
  }
  solution
}

# -- Tree walk helpers (plain functions for performance) -------------

## .canonicalize_tree: recursive bottom-up walk
## CVXPY SOURCE: canonicalization.py lines 86-123
.canonicalize_tree <- function(expr) {
  ## Recurse into each argument -- pre-allocate, flatten once
  n_args <- length(expr@args)
  canon_args <- vector("list", n_args)
  constr_chunks <- vector("list", n_args + 1L)
  for (i in seq_len(n_args)) {
    arg_result <- .canonicalize_tree(expr@args[[i]])
    canon_args[[i]] <- arg_result[[1L]]
    constr_chunks[[i]] <- arg_result[[2L]]
  }
  ## Canonicalize this node
  node_result <- .canonicalize_expr(expr, canon_args)
  constr_chunks[[n_args + 1L]] <- node_result[[2L]]
  constrs <- unlist(constr_chunks, recursive = FALSE)
  if (is.null(constrs)) constrs <- list()
  list(node_result[[1L]], constrs)
}

## .canonicalize_expr: canonicalize a single node
## CVXPY SOURCE: canonicalization.py lines 125-157
.canonicalize_expr <- function(expr, args) {
  ## Skip constants (no parameters) -- collapse them
  if (S7_inherits(expr, Expression) &&
      is_constant(expr) && length(parameters(expr)) == 0L) {
    return(list(expr, list()))
  }

  ## S7 dispatch -- default method returns identity copy
  dcp_canonicalize(expr, args)
}