inst/doc/v06-self-modify-pipeline.R

In pipeflow: Fast Interactive Data Analysis Pipelines

## ----knitr-setup, include = FALSE---------------------------------------------------------------------------------------------------------------------------------------------------------------------
require(pipeflow)

knitr::opts_chunk$set(
    comment = "#",
    prompt = FALSE,
    tidy = FALSE,
    cache = FALSE,
    collapse = TRUE
)

old <- options(width = 200L)

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip <- pip_new("my-pipeline") |>
    pip_add("init", \(xInit = 0) xInit) |>
    pip_add("f1", \(x = ~init) x + 1) |>
    pip_add("f2", \(x = ~f1) x + 2) |>
    pip_add("f3", \(x = ~f2) x + 3)

str(pip)

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
ls(pip$pipenv)

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip$pipenv$data # or pip_data(pip)

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip[["f2", "step"]] <- "my_pretty_f2" # updates downstream 'depends'

pip

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip[step %like% "f", "locked"] <- TRUE

pip

## ----error = TRUE-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
try({
pip[["init", "nodeId"]] <- 99L
})

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
dat <- pip$pipenv$data
dat[2, "step"] <- "new f1 name" # fails to update downstream 'depends'
dat[3, "nodeId"] <- 99L # breaks link to internal DAG node
assign("data", dat, envir = pip$pipenv)

pip

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip <- pip_new("my-pipeline") |>
    pip_add("init", \(xInit = 0) xInit) |>
    pip_add("f1", \(x = ~init) x + 1) |>
    pip_add("f2", \(x = ~f1) x + 2) |>
    pip_add("f3", \(x = ~f2) x + 3)

(pip_run(pip))

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip |> pip_replace(
    "f2",
    \(x = ~f1) {
        if (x > 10) {
            .self$replace("f3", \(x = ~f1) x * 3)
            return(x / 2)
        }
        x + 2
    }
)

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip |>
    pip_set_params(list(xInit = 15)) |>
    pip_run()

pip

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip[["f3", "fun"]]

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip <- pip_new("hicky-hacky") |>
    pip_add("init", \(xInit = 0) xInit) |>
    pip_add("f1", \(x = ~init) x + 1) |>
    pip_add(
        "f2",
        \(x = ~f1) {
            if (x > 10) {
                .self |>
                    pip_add("f2a", \(x = ~f1) x + 21, after = "f1") |>
                    pip_add("f2b", \(x = ~f2a) x + 22, after = "f2a") |>
                    pip_replace("f3", \(x = ~f2b) x + 30) |>
                    pip_remove("f2")
            }
            x + 2
        }
    ) |>
    pip_add("f3", \(x = ~f2) x + 3)

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip_run(pip)

pip

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip |>
    pip_set_params(list(xInit = 11)) |>
    pip_run()

pip

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip |>
    pip_set_params(list(xInit = 11)) |>
    pip_run()

pip

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip <- pip_new("hacky-with-restart") |>
    pip_add("init", \(xInit = 0) xInit) |>
    pip_add("f1", \(x = ~init) x + 1) |>
    pip_add(
        "f2",
        \(x = ~f1) {
            if (x > 10) {
                .self |>
                    pip_add("f2a", \(x = ~f1) x + 21, after = "f1") |>
                    pip_add("f2b", \(x = ~f2a) x + 22, after = "f2a") |>
                    pip_replace("f3", \(x = ~f2b) x + 30) |>
                    pip_remove("f2")
                .self$restart() # <-- restart the run
            }
            x + 2
        }
    ) |>
    pip_add("f3", \(x = ~f2) x + 3)

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip |>
    pip_set_params(list(xInit = 11)) |>
    pip_run()

## -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
pip