R/datapull.R

In mlhutchins/fast-analytics: Fast Prototyping of Analysis Scripts

#' The Data Pull Helper Functions
#'
#' All data pull texts are defined as standard functions that assume the csid
#' variable is called csid and con1 and con2 (for rsr-history and ndp) are defined)
#'
#' @return List of the data pull text and necessary database connection
#' @name dataPull
NULL

#' Pull Test Summary
#' @rdname dataPull
pullTestSummary <- function() {

    pull <- "
data <- do.call('rbind', lapply(csid, function(x) {
    loadCache(sprintf('%stest_summary_with_meta_%g.RData', cacheDir, x),
              fetchTestSummaryWithMeta, con1, x, rm = reprocess)
}))"
    return(pull)

}

#' Pull Base Map
#' @rdname dataPull
pullBaseMap <- function() {

    pull <- "
baseMap <- ggGoogle(data$lon, data$lat, source = 'stamen', maptype = 'toner')"

    return(pull)

}

#' Pull Call Statistics
#' @rdname dataPull
pullCallStats <- function() {

    pull <- "
callStats <- do.call('rbind', lapply(csid, function(x) {
    loadCache(sprintf('%scall_stats_%s.Rdata', cacheDir, x),
    	      fetchCallStats, con1, csid, 30000, rm = reprocess)
}))"

    return(pull)

}

#' Pull Poll Event
#' @rdname dataPull
pullPollEvent <- function() {

    pull <- "
poll <- do.call('rbind', lapply(csid, function(x) {
loadCache(sprintf('%sdevice_poll_event_%s.Rdata', cacheDir, x),
				  fetchPollWithLocation, con2, unique(data$device_id), range(data$device_time),
				  'device_poll_event', rm = reprocess)
}))"
    return(pull)

}

#' Pull Filtered Test Ssummary
#' @rdname dataPull
pullFilteredTestSummary <- function() {

    pull <- "
filtered <- do.call('rbind', lapply(csid, function(x) {
    loadCache(sprintf('%sfiltered_test_summary_%g.RData', cacheDir, x),
              fetchFilteredTestSummary, con1, x, rm = reprocess)
}))"
    return(pull)

}

#' Pull RootScores
#' @rdname dataPull
pullRootScores <- function() {

    pull <- "
rootscores <- do.call('rbind', lapply(csid, function(x) {
    loadCache(sprintf('%srootscores_%g.RData', cacheDir, x),
              function() {
                   scores <- fetchRootscoreResults(con1, x)
                   if (nrow(scores) == 0) {
                        warning('No server side rootscores, calculating locally')
                        scores <- bootscorer(data, version = 7, cores = cores)$summaryScores
                   }
                   return(scores)
            }, rm = reprocess)
}))"

    return(pull)

}

#' Pull Dataspeed Intervals
#' @rdname dataPull
pullDataspeed <- function() {

    pull <- "
dataspeed <- do.call('rbind', lapply(csid, function(x) {
    loadCache(sprintf('%sdataspeed_interval_%g.RData', cacheDir, x),
              function(){
                  meta <- getMetadata(con1, csid)
                  fetchDataspeed(con = con2, devices = unique(meta$device_id),
                                 date = c(min(meta$collection_time_start),
                                          max(meta$collection_time_end)))},
              rm = reprocess)
}))
"
    return(pull)

}

#' UDP By Size Class
#' @rdname dataPull
pullUDPExpanded <- function() {

    pull <- "
udpExpanded <- do.call('rbind', lapply(csid, function(x) {
    loadCache(sprintf('%sudp_expanded_%g.RData', cacheDir, x),
              fetchUDPExpanded, con1, x, rm = reprocess)
}))"

	return(pull)
}

#' UDP By Size Class Processing
#' @rdname dataPull
udpLatencyMelt <- function() {

    process <- "
udpLatencyMelt <- melt(udpExpanded,
		   id.vars = c('device_id', 'device_time', 'carrier', 'collection_set_id'),
		   measure.vars = c(grep('latency_median_',
		   					names(udpExpanded),
		   					value = TRUE), 'latency_median'))
"

    return(process)

}


#' UDP By Size Class Processing
#' @rdname dataPull
udpDropMelt <- function() {

    process <- "
udpDropMelt <- melt(udpExpanded,
		   id.vars = c('device_id', 'device_time', 'carrier', 'collection_set_id'),
		   measure.vars = c(grep('packet_drop_',
		   					names(udpExpanded),
		   					value = TRUE), 'avg_packet_drop'))
"

    return(process)

}