---
title: "pippy"
vignette: >
  %\VignetteIndexEntry{pippy}
  %\VignetteEngine{quarto::html}
  %\VignetteEncoding{UTF-8}
  %\VignetteDepends{fs}
knitr:
  opts_chunk:
    collapse: true
    comment: '#>'
    fig.width: 8
    fig.height: 5
    fig.align: 'center'
---

## Background

pippy is a package designed to assist with the running of analytical pipelines across multiple scenarios.
Different stages of analysis are defined as functions which are then brought together as a pipeline using a syntax that is a reduced subset of R.
This pipeline is then represented as a [directed acyclic graph](https://en.wikipedia.org/wiki/Directed_acyclic_graph) (DAG)] allowing us to employ robust approach to work planning and caching.
Whilst the goal is to be "good enough" for many pieces of analysis, by encouraging a functional approach, it remains easy to move to more advanced alternatives
such as [targets](https://docs.ropensci.org/targets/).

Goals:

- 🗸 Intelligent caching to reduce unnecessary execution.
- 🗸 Configuration files using a reduced subset of base R as their syntax.
- 🗸 Simple and efficient approach to managing multiple scenarios.

Out of scope:

- 🗴 HPC orchestration.
- 🗴 Automatic parallelisation of the pipeline. 
- 🗴 Environment management.

Caveats:

- For serialising objects to disk we utilise R's rds format (with the ability to pass arguments through).
  This means that functions should only return objects that are serialisable and unserialisable by that format.
  
- When working out which stages of a pipeline need rerunning we take in to account functions defined by the user in the function directory.
  We do not account for functions of called packages or base R and stats.
  This means that updating packages or R can potentially invalidate the cache so it is the users responsibility to maintain a consistent environment, or `force` a rerun of the pipeline where necessary.
  Future versions of pippy will warn when underlying package versions (or R itself) change between runs.

- The only global object we account for is `.Random.seed`.
  If you wish to take advantage of object caching then it is important to set a seed before running a pipeline.
  This seed will then be utilised at the start of each scenario run.


## A motivating example

The easiest way to get started with the package is to look at a template pipeline included within the package.
`create_skeleton_pipeline()` can be used to set up a pipeline based on the mtcars data set and multiple scenarios.

```{r}
#| eval: false
library(pippy)
dir <- create_skeleton_pipeline(dir = tempfile())
od <- setwd(dir)
```

```{r}
#| echo: false
library(pippy)
f <- tempfile()
dir <- tryCatch(
    create_skeleton_pipeline(f),
    error = function(cond) {
        unlink(f)
        stop("Unable to create pipeline skeleton.")
    }
)
```

We now have the following in `dir`:

```{r}
#| echo: false
if (requireNamespace("fs", quietly = TRUE)) {
    fs::dir_tree(dir)
} else {
    list.files(dir, all.files = TRUE, recursive = TRUE, no.. = TRUE)
}
```

The pipeline we are going to run is described in `pipeline.R`:

```{.r filename="pipeline.R"}
set.seed(1)
out <- run_pipeline(
    {
        raw   <- load_dat(CONFIG$in_csv)
        clean <- wrangle_dat(raw, CONFIG$rows)
        plot  <- plot_dat(clean, CONFIG$out_plot)
    },
    scenario = "production",
    scenario_default = "default"
)
```

`run_pipeline()` takes an expression where each assignment it contains represents a stage of the analysis.
There are a few things to note about the expression syntax:

- The left-hand-side (target) of each assignment must be uniquely named.

- Function calls on the right-hand-side (rhs) must be defined in the function directory `R` or contained in the base R or stats namespaces.

- The target names cannot match the names of functions in base R, or those contained in the function directory.

- The order of assignments is not important as it is determined internally by the function.

- Values from the configuration file, `config.R`, are accessed by the special `CONFIG` value (a list which can be subset using the `$` symbol).

The last two entries relate to the configuration file:

- `scenario_default` tells the function to parse the configuration file, looking for an entry called 'default'.
This is used as the base scenario.

- `scenario` (in this case 'production' but could be multiple ones) specifies other values in the configuration file that are then layered on top of this (internally using `modifyList()`) to generate additional scenarios to run.


### The configuration file

Configuration files are simply R scripts that are evaluated in a dedicated environment.

Scenarios are defined by named lists.

Within this environment we enable two special functions `INFILE` and `OUTFILE`:

- `INFILE()` marks a file as an input.
This ensures that `run_pipeline()` knows to keep a hash of the object and to monitor it for future changes when the pipeline is rerun.

- `OUTFILE()` marks a file name as an output. This ensures that outputs from different scenarios (see later) do not overwrite one another (output files get saved in a scenario-dependent directory). It also ensures that output files are recreated if they are deleted.

In our example we have the following configuration file:

```{.r filename="config.R"}
default <- list(
    in_csv   = INFILE("data/mtcars.csv"),
    out_plot = OUTFILE("plot.png"),
    rows     = 10
)

production <- list(
    rows = 32
)
```

Here `in_csv` points to our input data and `out_plot` to a png output file.
The 'production' scenario will then be the same as the default but with the number of rows selected being 32, not 10.

This time the configuration file defines two scenarios, *default*, which subsets the data to 10 rows and *production*, which returns 32 rows.


### The R directory

This is the directory where users define functions that represent stages of their analysis (e.g. cleaning data, modelling, plotting, ...).
Here we have:

```{.r filename="R/load_dat.R"}
load_dat <- function(file) {
    stopifnot(file.exists(file))
    # Note: If function not in the base/stats namespace or function directory
    #       we must explictly namespace the call using `::`.
    utils::read.csv(file)
}
```

This file contains a single function which takes the filename of raw data as input before doing a little wrangling and returning the output.
Note the comment about namespacing functions;
anything not in the base/stats namespace or the function directory needs to be explicitly namespaced.

Next we have:

```{.r filename="R/wrangle_dat.R"}
wrangle_dat <- function(dat, rows) {
    stopifnot(is.data.frame(dat), nrow(dat) >= rows, rows >= 0)
    head(dat, rows)
}
```

This function curtails the number of rows of data.
Here we also make use of assertions in the function,
checking that `dat` is a data frame and `rows` is between 0 and the number of rows in the data frame (inclusive).
Assertions like these can be thought of unit tests for your analysis. 
Whilst here we have them at the start of the function to check inputs,
they can be used anywhere in your function to ensure things are running as expected.
If an assertion fails it will stop the pipeline running, bringing your attention to any issue at hand.

Finally there is

```{.r filename="R/plot.R"}
plot_dat <- function(dat, outfile) {
    grDevices::png(outfile)
    plot(factor(dat$cyl), dat$mpg, xlab = "cyl", ylab = "mpg")
    grDevices::dev.off()
    # return name of where image is saved
    outfile
}

```

This function takes the manipulated data as input and creates a plot which is saved as a png.
When saving output to disk, we recommend that the function returns the location where the file is saved for later use.


### Putting it together

Running the pipeline we see the following

```{r}
#| eval: false
set.seed(1)
out <- run_pipeline(
    {
        raw <- load_dat(CONFIG$in_csv)
        clean <- wrangle_dat(raw, CONFIG$rows)
        plot <- plot_dat(clean, CONFIG$out_plot)
    },
    scenario = c("default", "production"),
    scenario_default = "default" # included here but note that this is
                                 # also the default value for the argument.
)
```

```{r}
#| echo: false
set.seed(1)
out <- run_pipeline(
    {
        raw <- load_dat(CONFIG$in_csv)
        clean <- wrangle_dat(raw, CONFIG$rows)
        plot <- plot_dat(clean, CONFIG$out_plot)
    },
    scenario = c("default", "production"),
    scenario_default = "default",
    pipeline_dir = dir
)
```

We see above that the *default* scenario ran from scratch where as the *production* scenario was able to skip the first computation, loading the cached result from disk.
This caching can also takes account of changes in functions within the function directory but does not attempt to handle changes in the underlying R version or packages you use 
(in essence, it assumes environment stability is managed elsewhere).

Each target output is saved within a named list associated with the relevant scenario.

```{r}
str(out)
str(out$default$clean)
str(out$production$clean)
out$default$plot
out$production$plot
```

```{r}
#| echo: false
#| fig.cap: "The resulting plots in the 'output' directory."
#| fig.alt: "Two side by side box plots of cyl (x-axis) versus mpg (y-axis). The left-hand one represents the 'default' configuration, the right-hand, 'production'. There is a downward trend showing in mpg as cyl increases." # nolint: line_length_linter.
layout(matrix(c(1, 2), nrow = 1))
ylim <- c(0, max(out$production$clean$mpg) + 2)
plot(
    factor(out$default$clean$cyl), out$default$clean$mpg,
    ylim = ylim, xlab = "cyl", ylab = "mpg", main = "default"
)
plot(
    factor(out$production$clean$cyl), out$production$clean$mpg,
    ylim = ylim, xlab = "cyl", ylab = "mpg", main = "production"
)
```


### Changing things

Let's assume that we decide the `rows` value for our production run is changed to 30:


```{r}
#| eval: false
cat(
    "\nproduction <- list(rows = 30)\n",
    file = "config.R",
    append = TRUE
)
```
	
```{r}
#| echo: false
cat(
    "\nproduction <- list(rows = 30)\n",
    file = file.path(dir, "config.R"),
    append = TRUE
)
```

```{r}
#| eval: false
set.seed(1)
out <- run_pipeline(
    {
        raw     <- load_dat(CONFIG$in_csv)
        clean   <- wrangle_dat(raw, CONFIG$rows)
        plot    <- plot_dat(clean, CONFIG$out_plot)
    },
    scenario = "production",
    scenario_default = "default"
)
```

```{r}
#| echo: false
set.seed(1)
out <- run_pipeline(
    {
        raw     <- load_dat(CONFIG$in_csv)
        clean   <- wrangle_dat(raw, CONFIG$rows)
        plot    <- plot_dat(clean, CONFIG$out_plot)
    },
    scenario = "production",
    scenario_default = "default",
    pipeline_dir = dir
)
```

This time our caching ensured that only the production scenario was rerun and that the cached raw object was not recalculated.


### Loading objects from the cache

For large output, it can be beneficial not to return everything.
To this end we provide a `return` argument to stop a running pipeline from returning it's output.
Output can then be accessed with `load_pipeline_object`:

By default all scenarios are returned:

```{r}
#| eval: false
load_pipeline_object("clean")
```

```{r}
#| echo: false
load_pipeline_object("clean", pipeline_dir = dir)
```

We can restrict scenarios if we wish:

```{r}
#| eval: false
load_pipeline_object("clean", scenario = "default")
```

```{r}
#| echo: false
load_pipeline_object("clean", scenario = "default", pipeline_dir = dir)
```

Finally we go back to our original directory

```{r}
#| eval: false
setwd(od)
```