# iRadioMeta.r if (FALSE) { ## SOURCE THIS FILE source("~/git/orch/src/iRadio/iRadioMeta.r") ## Grab all itunes Tables.. (Excep itunes_iradio_report, which is blocked) tbls.itunes <- qShowTables("itunes")[tablename != "itunes_iradio_report", tablename] ## Set names to self, so that lapply grabs them setattr(tbls.itunes, "names", tbls.itunes) ## Pull the head of each table heads.itunes <- lapply(tbls.itunes, headDB, 25) } ### TODO: Put all of the .meta.* lists into their own environment, then cleaniRadioCols should pull from that environment. cleaniRadioCols_ <- function(DT, type.of.iradio.data.source=c("event", "customer", "station", "content") , add.time.info=TRUE, drop.these.columns=NULL, confirm.drop=FALSE , verbose=TRUE, showWarnings=verbose) { # type.of.data.source should be any combination of the the available options ## We will have one .meta.iradio... list for each combination of type-class, eg ## .meta.iradio.type.class ## where 'type' refers to the type of *source* for the data ## and 'class' refers to the *expected* (output) class of the column hr <- pasteR(80) verboseMsg(verbose, "\n", time=FALSE) verboseMsg(verbose, " WILL GATHER COLUMNS BY META TYPTE TO SEND TO fixColTypes_() \n", hr, time=TRUE) col_classes <- c("idcol", "POSIXct.milli", "factor_list", "notUsing", "character") for (cl in col_classes) { .m.nm <- sprintf(".meta.iradio.%s.%s", type.of.iradio.data.source, cl) ## Get all of the meta info, col names z <- lapply(.m.nm, function(x) if (exists(x)) get(x)) ## TODO: Add the Environment ## Unlist the lapply results, making sure not to recurse (eg factor_list needs to be a list) ret <- unlist(z, recursive=FALSE, use.names=TRUE) ## Take unique values. Using unique() will drop attributes, thus use x[!duplicated(x)] ret <- ret[!duplicated(ret)] ## make sure they are in names(DT) { if (is.list(ret)) ret <- ret[names(ret) %in% names(DT)] else ret <- ret[ret %in% names(DT)] } ## Assign it! assign(cl, ret) verboseMsg(verbose, sprintf("Collected %-15s which has %2i items", cl, length(ret)), time=FALSE) } ## sapply preserves name, which will be needed in do.call (lapply does not preserve name) args <- sapply(col_classes, function(x) {if (exists(x) && length(get(x))) get(x) }, simplify=FALSE) args <- c(DT="DT", args, verbose=verbose) # Clean this one .meta. name names(args)[names(args) == "notUsing"] <- "drop.these.columns" ## User must set this flag to TRUE to actually drop columns if (!confirm.drop) { warning ("Unused columns will NOT be dropped.\nPlease set confirm.drop=TRUE to drop them.") args[["drop.these.columns"]] <- NULL } else { ## Also add in any 'drop.these.columns' which are in the names(DT) drop.these.columns <- intersect(drop.these.columns, names(DT)) args[["drop.these.columns"]] <- c(args[["drop.these.columns"]], drop.these.columns) ## Verbose output. Notify user which columns, if any will be dropped if (length(args[["drop.these.columns"]])) verboseMsg(verbose, "Will be dropping the following columns from DT: ", args[["drop.these.columns"]], time=FALSE) else verboseMsg(verbose, "No columns are selected for dropping (ie any unneeded columns were already excluded)", time=FALSE) } ## EXECUTE verboseMsg(verbose, "\n", time=FALSE) verboseMsg(verbose, " CALLING fixColTypes_() \n",hr, time=TRUE) do.call(fixColTypes_, args) ## CLEAN COLUMN NAMES dict.nms <- c( ## OLD = ## NEW customer_identifier = "customerid" , apple_identifier = "appleid" , station_id = "stationid" , storefront_name = "storefront" , station_name = "station" , extracted_upc_long = "upc" , extracted_upc = "upc" , title = "track" , event_start_time = "start_time" , event_end_time = "end_time" ) dict.nms <- dict.nms[names(dict.nms) %in% names(DT)] setnames(DT, names(dict.nms), dict.nms) ## We will calculate duration by either appleid if exists, else isrc duration.bycol <- ifelse("appleid" %in% names(DT), "appleid", "isrc") if (add.time.info && duration.bycol %in% names(DT)) { ## Add DayOfWeek DT[, day := format(start_time, "%a")] ## Add Duration DT[, duration := as.integer(round(end_time - start_time))] ## Add Duration percentage DT[, duration.perc := round(duration / quantile(duration, 0.99), 4), by=duration.bycol] ## Add a 30-second marker, in terms of percentage DT[, ThirtySecMark := 30 / duration] } return(invisible(DT)) } # fixColTypes_ (DB.iradio.customer # , idcol = .meta.iradio.customer.idcol # , factor_list = .meta.iradio.customer.factor_list # , drop.these.columns = .meta.iradio.customer.notUsing) # # fixColTypes_ (DB.iradio.station # , idcol = .meta.iradio.station.idcol # , character = .meta.iradio.station.character # , factor_list = .meta.iradio.station.factor_list # , drop.these.columns = .meta.iradio.station.notUsing) # # fixColTypes_ (DB.iradio.content # , idcol = .meta.iradio.content.idcol # , factor = .meta.iradio.content.factorcols # , drop.these.columns = .meta.iradio.content.notUsing) .origin <- as.Date("1970-01-01", format="%Y-%m-%d") .origin.utc <- as.POSIXct("1970-01-01 00:00:00.000", format="%Y-%m-%d %H:%M:%S", tz="UTC") .meta.iradio.event.factor_list <- list( 'end_reason' = list( levels = 0:2 , labels = c("Natural", "Skipped", "Other") ) , 'event_type' = list( levels = 1:6 , labels = c("Played", "2 (undefined)", "Liked", "Banned", "Wished", "Unwished") ) , 'device_type' = list( levels = c("M", "D") , labels = c("Mobile", "Desktop") ) , 'track_ownership' = list( levels = 0:4 , labels = c("Not Applicable", "Not Owned", "Owned", "Owned & in Genius", "Owned & in Match") ) ) .meta.iradio.event.POSIXct.milli <- c("event_start_time", "event_end_time") .meta.iradio.event.idcol <- c("customer_identifier", "apple_identifier", "station_id", "storefront_name") .meta.iradio.event.notUsing <- c("ingestion_time", "filename", "download_date") ## This is the statement to call when ready to clean ## ie: ## eval(.meta.iradio.event.FIXCOLS) .meta.iradio.event.FIXCOLS <- quote(fixColTypes_(DB.iradio.event, idcol=.meta.iradio.event.idcol, POSIXct.milli=.meta.iradio.event.POSIXct.milli, factor_list=.meta.iradio.event.factor_list, drop.these.columns=.meta.iradio.event.notUsing)) # -------------------------------------------------------------------------------- # .meta.iradio.customer.factor_list <- list( 'user_type' = list( levels = c("M", "") , labels = c("Match Subscriber", "Not a Match Subsc") ) ) .meta.iradio.customer.idcol <- c("customer_identifier", "postal_code") .meta.iradio.customer.notUsing <- c("ingestion_time", "filename", "download_date") ## This is the statement to call when ready to clean ## ie: ## eval(.meta.iradio.customer.FIXCOLS) .meta.iradio.customer.FIXCOLS <- quote(fixColTypes_(DB.iradio.customer , idcol = .meta.iradio.customer.idcol , factor_list = .meta.iradio.customer.factor_list , drop.these.columns = .meta.iradio.customer.notUsing) ) # -------------------------------------------------------------------------------- # .meta.iradio.station.factor_list <- list( 'station_type' = list( levels = c("P", "U") , labels = c("Program", "User Created") ) ) .meta.iradio.station.idcol <- c("station_id") .meta.iradio.station.character <- c("station_id") .meta.iradio.station.notUsing <- c("ingestion_time", "filename", "download_date") ## This is the statement to call when ready to clean ## ie: ## eval(.meta.iradio.station.FIXCOLS) .meta.iradio.station.FIXCOLS <- quote(fixColTypes_(DB.iradio.station , idcol = .meta.iradio.station.idcol , character = .meta.iradio.station.character , factor_list = .meta.iradio.station.factor_list , drop.these.columns = .meta.iradio.station.notUsing) ) # -------------------------------------------------------------------------------- # # ---------------------------------------------------------------------------------------------- # .meta.iradio.content.idcol <- c("apple_identifier", "isrc", "vendor_identifier", "vendor_offer_code", "extracted_upc") .meta.iradio.content.factorcols <- c("artist", "title", "label") .meta.iradio.content.notUsing <- c("ingestion_time", "filename", "download_date", "extracted_upc", "extracted_isrc") ## This is the statement to call when ready to clean ## ie: ## eval(.meta.iradio.content.FIXCOLS) .meta.iradio.content.FIXCOLS <- quote(fixColTypes_(DB.iradio.content , idcol = .meta.iradio.content.idcol , factor = .meta.iradio.content.factorcols , drop.these.columns = .meta.iradio.content.notUsing) ) # -------------------------------------------------------------------------------- # # ---------------------------------------------------------------------------------------------- # merge_iRadio <- function(base.nm, event=TRUE, station=TRUE, customer=TRUE, content=TRUE, clean.up.after=TRUE, verbose=TRUE, debug=FALSE) { if (!event) stop ("event is the main DT. Withouth it, you must manually merge.") ## Verbose output start.time <- proc.time() hr <- pasteR(80) verboseMsg(verbose, "\n", time=FALSE) cols.using <- list( event = c("device_type", "track_ownership", "storefront_name", "event_type", "end_reason", "event_start_time", "event_end_time", "customer_identifier", "station_id") , station = c("station_type", "station_name") , customer = c("postal_code", "user_type") , content = c("artist", "title", "isrc", "extracted_upc") ) # MERGE bycols.using <- list( event = c("apple_identifier") , station = c("station_id") , customer = c("customer_identifier") , content = c("apple_identifier") ) nms.using <- c("event"[event], "station"[station], "customer"[customer], "content"[content]) # Filter out those not using cols.using <- cols.using [nms.using] bycols.using <- bycols.using [nms.using] ## Concat the names of the DT's to geT() DT.nms <- paste0(base.nm, nms.using) ## Check that all exist DT.nms <- setNames(nm=DT.nms) if (length(miss <- nwhich(!sapply(DT.nms, exists)))) stop (warningCols("The following tables could not be found: ", miss)) ## Set the names to just the basic "event", "content" etc setattr(DT.nms, "names", nms.using) ## Duplicate-resolution Function ## When ambiguous, we will take the longest value longest <- function(x) { if (length(x) > 1) x[which.max(nchar(x))] [[1]] else x } ## Function to drop any columns that are not present in the DT colNameIsinDT <- function(col.list) { ## TODO: Do we want to warn on this being filtered? mapply(function(nm, cols) cols[cols %in% names(get(nm))], DT.nms, col.list) } cols.using <- colNameIsinDT(cols.using) bycols.using <- colNameIsinDT(bycols.using) if (any(wh <- 0 == lapply(bycols.using, length))) stop ("by.cols for ", pasteQ(DT.nms[wh]), " is empty") browser(expr=isTRUE(debug), text="in merge_iRadio(), before the mapply clause") #### FOR DEBUGGING ## i <- 1; x=DT.nms[[i]]; cols=cols.using[[i]]; bycols=bycols.using[[i]]; nm=nms.using[[i]]; ## Pull each table. Take unique rows. Then in each column, resolve duplicates, by id, using longest() DTs <- mapply(function(x, cols, bycols, nm) { verboseMsg(verbose, sprintf("Processing & uniquifying '%-20s", paste0(nm, "' table"))) if (nm == "event") unique(get(x)[, c(bycols, cols), with=FALSE], by=NULL) ## don't uniquify the columns else unique(get(x)[, c(bycols, cols), with=FALSE], by=NULL) [, lapply(.SD, longest), by=bycols] } , DT.nms, cols.using, bycols.using, nms.using) ## Ensure that 'event' table does not contain any duplicates if (any(duplicated(DTs[["event"]]))) stop("Duplicates in the event table") #### -= START MERGING =- #### DT <- DTs[["event"]] for (nm in nms.using[!nms.using=="event"]) { DT <- merge(DT, DTs[[nm]], by=bycols.using[[nm]]) } ## CONFIRMATION ## Check that all the columns we were planning to use have made it in if (any(miss <- !unlist(c(cols.using, bycols.using)) %in% names(DT))) warning(warningCols("These columns did not make it into the merge", unlist(c(cols.using, bycols.using))[miss] )) ## CLEANUP if (clean.up.after && exists("cleaniRadioCols_")) cleaniRadioCols_(DT, verbose=verbose, confirm=TRUE) total.time <- {proc.time() - start.time} [["elapsed"]] verboseMsg(verbose, "\n", hr, "\n\tTOTAL PROCESSING TIME WAS ", fwSecs(total.time), "\n", hr, sep="", time=FALSE) return(DT) } # -------------------------------------------------------------------------------- # # ---------------------------------------------------------------------------------------------- #