R/q.run.R

In okux/qrmarkdown: Job Queue Management For Rmarkdown

q.run<-function(jid)
{
  require(rmarkdown)
  check <- Sys.getenv("RSTUDIO_PANDOC")
  if(length(check) == 0) stop("define RSTUDIO_PANDOC path!")

  SIGNAL <- TRUE

  job.run <- function(ticket_path)
  {

    # Loading Ticket
    tryCatch({
      if ( !file.exists(ticket_path) )
        stop( paste('error ticket not found', ticket_path))

      load(file=ticket_path)
      SIGNAL<-FALSE
      print(ticket$name)
      print(ticket$params)

    }, error=function(e){})


    if (SIGNAL == FALSE)
    {
      tryCatch({
        # update ticket status
        ticket$jid    <- q.move(ticket_path, to = q.status$running)
        ticket$status <- q.status$running
        ticket$ctime  <- Sys.time()
        ticket$secs   <- as.integer(ticket$ctime - ticket$mtime)

        save(ticket, file = ticket$jid)

        #setup markdown runtime diretory
        OUTDIR <-
          sprintf('%s/%s', .pkg.log$output.dir, basename(ticket$jid))
        dir.create(OUTDIR, recursive = TRUE, showWarnings = FALSE)

        OUTFILE <-
          sprintf('%s/%s.html', OUTDIR, basename(ticket$script))
        OUTFILE <- gsub('.Rmd', '', OUTFILE)

        # adding dynamic parameter feature to get rid
        # kinitr YALM param issues

        if (ticket$jparams)
          RMD_PARAMS <- list(jid = ticket$jid)
        else
          RMD_PARAMS <- ticket$params

        if (!is.null(ticket$output.file))
        {
          # RUN
          out <- rmarkdown::render(
            input = ticket$script,
            output_file = OUTFILE,
            output_dir = OUTDIR,
            intermediates_dir = OUTDIR,
            # fix concurrent
            params = RMD_PARAMS
          )

          # copy to specific location
          if (file.exists(ticket$output.file))
            file.remove(ticket$output.file)

          file.copy(from = out,
                    to = ticket$output.file,
                    overwrite = TRUE)
          file.remove(out)
          message("Your output file:\t", ticket$output.file)
        } else{
          out <- rmarkdown::render(
            input = ticket$script,
            output_file = OUTFILE,
            output_dir = OUTDIR,
            intermediates_dir = OUTDIR,
            params = RMD_PARAMS
          )

          ticket$output.file <- out
        }

        # update success status
        ticket$status <- q.status$completed
        ticket$secs   <-
          as.integer(Sys.time() - ticket$ctime) # calcurate total run time
        ticket$ctime  <- Sys.time()

        ticket$jid     <- q.move(ticket$jid, to = 'outbox')
        save(ticket, file = ticket$jid)

        ret <- data.frame(
          jid = ticket$jid,
          status = ticket$status,
          secs = ticket$secs,
          output = ticket$output.file
        )

        message(ticket$jid)
        return(ret)
      },
      error = function(e)
      {
        # update error status
        ticket$status <- q.status$failed
        ticket$secs   <- as.integer(Sys.time() - ticket$ctime)
        ticket$ctime  <- Sys.time()
        ticket$output.file = "#ERROR#"
        ticket$log = gsub('\n', '', as.character(e))

        ticket_path   <- q.move(ticket$jid, to = q.status$failed)
        ticket$jid    <- ticket_path

        save(ticket, file = ticket_path)
        ret <- data.frame(
          jid = ticket$jid,
          status = ticket$status,
          secs = ticket$secs,
          output = ticket$output.file, stringsAsFactors = FALSE
        )

        message(ticket$jid)
        return(ret)
      })

    }


  }

  # Handling multiple ticket(s)
  if(is.vector(jid))
  {
    return ( bind_rows (lapply(jid, job.run)) )
  } else return ( job.run(jid) )

}