R/helper.R

In numeract/rflow: Flexible R Pipelines with Caching

# helper functions not present in nmisc.R


# !diagnostics suppress=.,


is_key <- function(x) {
    
    if (typeof(x) != "character") return(FALSE)
    if (length(x) != 1L) return(FALSE)
    if (x %in% c(NA_character_, "")) return(FALSE)
    
    TRUE
}


require_keys <- function(...) {
    
    key_lgl <- purrr::map_lgl(list(...), ~ is_key(.))
    
    if (sum(!key_lgl) > 0L) {
        rlang::abort("key arguments must be valid strings")
    }
}


make_key <- function(fn_name, fn, fn_id, flow_options, class_name) {
    
    # fn_name
    stopifnot(!is.null(fn_name))
    if (!is.character(fn_name)) {
        if (is.symbol(fn_name)) {
            fn_name <- as.character(fn_name)
        } else if (is.language(fn_name)) {
            rlang::abort("Anonymous functions not supported.")
        } else {
            rlang::abort("Unrecognized `fn_name` data type.")
        }
    }
    require_keys(fn_name, class_name)
    # fn
    stopifnot(is_not_flow_fn(fn))
    # fn_id
    if (is.null(fn_id)) {
        is_default_id <- TRUE
    } else {
        is_default_id <- FALSE
        stopifnot(is_key(fn_id) || (
            rlang::is_scalar_integerish(fn_id) 
            && is.finite(fn_id) && (fn_id >= 1))
        )
        if (rlang::is_scalar_integerish(fn_id)) fn_id <- as.integer(fn_id)
    }
    # when generated by get_flow_options(), flow_options are valid
    if (!is.null(flow_options$eval_arg_fn)) {
        fn_formals <- formals(args(fn))
        eval_arg_fn_formals <- formals(args(flow_options$eval_arg_fn))
        stopifnot(identical(fn_formals, eval_arg_fn_formals))
    }
    # class_name
    require_keys(class_name)
    
    # useful misc 
    eddy <- flow_options$eddy
    stopifnot(inherits(eddy, "R6Eddy"))
    # fn_name can be anything & exists only in R6Flow obj (not saved in cache)
    fn_names <- purrr::map_chr(eddy$flow_lst, "fn_name")
    
    # flow_hash: all but fn_name (always ignored) & fn_id; calc only once
    fn_formals <- formals(args(fn))
    arg_chr <- paste(
        paste(names(fn_formals), as.character(fn_formals), sep = "="),
        collapse = ", ")
    body_chr <- as.character(body(fn))
    if (length(body_chr) == 0L) {
        body_chr <- format(fn)
    }
    if (any(grepl("\\.Primitive", body_chr))) {
        rlang::abort("Primitive functions not supported.")
    }
    fo_chr <- format(discard_at(flow_options, "eddy"))
    flow_hash <- eddy$digest(c(arg_chr, body_chr, fo_chr, class_name))
    
    # if fn_id given ==> simple case, otherwise do some guessing
    if (is_default_id) {
        # get a list of values to try for fn_id (taking hint from fn_name)
        fn_keys <- names(fn_names %if_in% fn_name)
        fn_id_lst <- 
            eddy$flow_lst[fn_keys] %>%
            purrr::map("fn_id")
        fn_key_lst <-
            fn_id_lst %>%
            purrr::map(~ eddy$digest(c(flow_hash, .))) %>%
            purrr::imap_lgl(~ .x == .y) %>% 
            purrr::keep( ~ .) %>% 
            names()
        
        if (length(fn_key_lst) > 1L) {
            rlang::abort(paste0(
                "Found multiple flows; please supply `fn_id`."))
        } else if (length(fn_key_lst) == 1L) {
            fn_key <- fn_key_lst[[1L]]
            fn_id <- fn_id_lst[[fn_key]]
            action = "get"
            rlang::inform(paste0(
                "Reusing cache: fn=", fn_name, 
                " / fn_id=", fn_id, " / fn_key=", fn_key))
        } else {
            # try fn_id = 1L & check all keys; ignore fn_name
            fn_keys <- names(fn_names)
            fn_id <- 1L
            fn_key <- eddy$digest(c(flow_hash, fn_id))
            if (fn_key %in% fn_keys) {
                action = "get"
                rlang::inform(paste0(
                    "Reusing cache: fn=", fn_name, 
                    " / fn_id=", fn_id, " / fn_key=", fn_key))
            } else {
                action = "new"
                rlang::inform(paste0(
                    "New cache: fn=", fn_name, 
                    " / fn_id=", fn_id, " / fn_key=", fn_key))
            }
        }
    } else {
        # fn_id is given: we do not care about fn_name, do not print messages
        fn_keys <- names(fn_names)
        fn_key <- eddy$digest(c(flow_hash, fn_id))
        if (fn_key %in% fn_keys) {
            action = "get"
        } else {
            action = "new"
        }
    }
    
    list(
        action = action,    # string: 'new' || 'get'
        fn_key = fn_key,    # key
        fn_name = fn_name,  # key
        fn_id = fn_id       # string/key or positive integer
    )
}


parse_call <- function(pos = 2L) {
    
    # the full call into the caller of this function
    parent_call <- match.call(
        definition = sys.function(-1L),
        call = sys.call(-1L),
        expand.dots = TRUE,
        envir = parent.frame(3L)
    )
    
    token <- parent_call[[pos]]
    if (is.symbol(token)) {
        if (as.character(token) == "." && pos == 2L) {
            # assume . from %>% ==> hack
            rlang::abort("Pipelines %>% not yet supported.")
            unmatched_fn_call <- parent_call[[3L]]
            parent <- parent.frame()
            unmatched_fn_call[[2L]] <- parent[["fn_call"]]
            # TODO: %>% still does not work
            # parent[["fn_id"]] <- parent[["flow_options"]]
            # parent[["flow_options"]] <- get_flow_options()
        } else {
            rlang::abort("The first argument must be a function call.")
        }
    } else if (is.language(token)) {
        # un-matched argument of the parent call
        # this is the as.is call to function (and its arguments) to be flow-ed
        unmatched_fn_call <- token
    } else {
        rlang::abort("Unrecognized argument type, expected a function call.")
    }
    
    fn <- eval(unmatched_fn_call[[1L]])
    if (!is.function(fn)) {
        format_fn_call <- paste(format(unmatched_fn_call), collapse = " ")
        rlang::abort(paste("Not a function call:", format_fn_call))
    }
    if (any(grepl("\\.Primitive", format(fn)))) {
        rlang::abort("Primitive functions not supported.")
    }
    
    # match.call for the function (and its arguments) to be flow-ed
    fn_call <- match.call(
        definition = fn,
        call = unmatched_fn_call,
        expand.dots = TRUE,
        envir = parent.frame(3L)
    )
    
    fn_call
}