R/utils-classes.R

In bcgov/bcdata: Search and Retrieve Data from the BC Data Catalogue

# Copyright 2019 Province of British Columbia
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and limitations under the License.

## Add "bcdc_promise" class
as.bcdc_promise <- function(res) {
  structure(res,
            class = c("bcdc_promise", setdiff(class(res), "bcdc_promise"))
  )
}

as.bcdc_sf <- function(x, query_list, url, full_url) {
  structure(x,
            class = c("bcdc_sf", setdiff(class(x), "bcdc_sf")),
            query_list = query_list,
            url = url, full_url = full_url, time_downloaded = Sys.time())
}


as.bcdc_query <- function(x) {
  structure(x,
            class = c("bcdc_query", setdiff(class(x), "bcdc_query"))
  )
}


# print methods -----------------------------------------------------------

#' @export
print.bcdc_promise <- function(x, ...) {

  x$query_list$CQL_FILTER <- finalize_cql(x$query_list$CQL_FILTER)

  if (is.null(x$query_list$count)) {
    query_list <- c(x$query_list, count = 6) ## only add if not there.
  } else {
   query_list <- x$query_list
  }

  cli <- x$cli
  cc <- cli$post(body = query_list, encode = "form")

  catch_wfs_error(cc)

  ## pagination printing
  number_of_records <- bcdc_number_wfs_records(x$query_list, x$cli)
  chunk_size <- check_chunk_limit()

  if (!is.null(x$query_list$count)) {
    # head or tail have updated the count
    number_of_records <- x$query_list$count
  }

  parsed <- bcdc_read_sf(cc$parse("UTF-8"))
  fields <- ncol(parsed) - 1

  # Check if this was called using a whse name directly without going
  # through a catalogue record so don't have this info
  name <- ifelse(is_record(x$record),
                 paste0("'", x[["record"]][["name"]], "'"),
                 paste0("'", x[["query_list"]][["typeNames"]], "'"))
  cat_line_wrap(glue::glue("Querying {col_red(name)} record"))

  cat_bullet(strwrap(glue::glue("Using {col_blue('collect()')} on this object will return {col_green(number_of_records)} features ",
                        "and {col_green(fields)} fields")))
  if (number_of_records > chunk_size) { # this triggers pagination
    cat_bullet(strwrap(glue::glue("Accessing this record requires pagination and will make {col_green(ceiling(number_of_records/chunk_size))} separate requests to the WFS. ",
                                              "See ?bcdc_options")))
  }

  cat_bullet(strwrap("At most six rows of the record are printed here"))
  cat_rule()
  print(parsed)
  invisible(x)
}

#' @export
print.bcdc_record <- function(x, ...) {
  cat_line_wrap(cli::col_blue(cli::style_bold("B.C. Data Catalogue Record: ")), x$title)
  cat_line_wrap(cli::col_blue(cli::style_italic("Name: ")), x$name, " (ID: ", x$id, ")")
  cat_line_wrap(cli::col_blue(cli::style_italic("Permalink: ")), paste0(catalogue_base_url(), "dataset/", x$id))
  cat_line_wrap(cli::col_blue(cli::style_italic("Licence: ")), x$license_title)
  cat_line_wrap(cli::col_blue(cli::style_italic("Description: ")), x$notes)


  tidy_resources <- bcdc_tidy_resources(x)
  avail_res <- tidy_resources[tidy_resources$bcdata_available, , drop = FALSE]
  cat_line_wrap(cli::col_blue(cli::style_italic("Available Resources (", nrow(avail_res), "):")))
  cli::cat_line(" ", seq_len(nrow(avail_res)), ". ", avail_res$name, " (", avail_res$format, ")")

  cat_line_wrap(cli::col_blue(cli::style_italic("Access the full 'Resources' data frame using: ")),
                cli::col_red("bcdc_tidy_resources('", x$id, "')"))

  if ("wms" %in% formats_from_record(x)) {
    cat_line_wrap(cli::col_blue(cli::style_italic("Query and filter this data using: ")),
                  cli::col_red("bcdc_query_geodata('", x$id, "')"))
  }

  invisible(x)
}

record_print_helper <- function(r, n, print_avail = FALSE) {
  cat_line_wrap(n, ") ", clean_wfs(r$name))
  #cat_line_wrap("    description:", r$description)
  cat_line_wrap("format: ", clean_wfs(formats_from_resource(r)), indent = 3)
  if (r$format != "wms") cat_line_wrap("url: ", r$url, indent = 3)
  cat_line_wrap("resource: ", r$id, indent = 3)
  if (print_avail) {
    cat_line_wrap("available in R via bcdata: ",
        if (r$format == "zip") {
          "Will attempt - unknown format (zipped)"
        } else {
          r$bcdata_available
        })
  }
  if (r$bcdata_available)
    cat_line_wrap("code: ", "bcdc_get_data(record = '", r$package_id,
        "', resource = '",r$id,"')", indent = 3)
  cat_line()
}

#' @export
print.bcdc_recordlist <- function(x, ...) {

  len <- length(x)

  if (len == 0L) {
    cat_line_wrap("Your search returned no results. Please try a different query.")
    return(x)
  }

  cat_line_wrap("List of B.C. Data Catalogue Records")
  n_print <- min(50, len)
  cat_line_wrap(cli::col_blue("Number of records: ", len))
  if (n_print < len) {
    cat_line_wrap(cli::col_blue("Showing the top 50 results. You can assign the output of bcdc_search, to an object and subset with `[` to see other results in the set."))
  cat_line("")
    }
  cat_line_wrap("Titles:")
  x <- purrr::set_names(x, NULL)

  purrr::imap(unclass(x)[1:n_print], ~ {

    if (!nrow(bcdc_tidy_resources(x[[.y]]))) {
      cat_line_wrap(.y, ": ",purrr::pluck(.x, "title"))
      cat_line_wrap("This record has no resources. bcdata will not be able to access any data.", col = "red")
    } else {
      cat_line_wrap(.y, ": ",purrr::pluck(.x, "title"),
                    " (", paste0(unique(formats_from_record(.x)), collapse = ", "),
                    ")")
    }

    cat_line_wrap("ID: ", purrr::pluck(.x, "id"), indent = 1, exdent = 2)
    cat_line_wrap("Name: ", purrr::pluck(.x, "name"), indent = 1, exdent = 2)
  })

  cat_line()
  cat_line_wrap("Access a single record by calling `bcdc_get_record(ID)`
      with the ID from the desired record.")
  invisible(x)
}

#' @export
print.bcdc_group <- function(x, ...) {

  cat_line_wrap(cli::col_blue(
    cli::style_italic(
      "Group Description: "
    )
  ), unique(attr(x, "description")))



  cat_line_wrap(cli::col_blue(
    cli::style_italic("Number of datasets: ")), nrow(x))

  print(tibble::as_tibble(x))
}


#' @export
print.bcdc_query <- function(x, ...) {

  cat_line("<url>")
  if (length(x$url) > 1) {
    for (i in seq_along(x$url)) {
      cat_line_wrap(glue::glue("Request {i} of {length(x$url)} \n{x$url[i]}"))
    }
  }

  cat_line("<body>")
  cat_line_wrap(glue::glue("   {names(x$query_list)}: {x$query_list}"))
  cat_line()
  cat_line("<full query url>")
  cat_line_wrap(x$full_url)
  invisible(x)
}


# dplyr methods -----------------------------------------------------------


#' Filter a query from bcdc_query_geodata()
#'
#' Filter a query from Web Feature Service using dplyr
#' methods. This filtering is accomplished lazily so that
#' the full sf object is not read into memory until
#' `collect()` has been called.
#'
#' @param .data object of class `bcdc_promise` (likely passed from [bcdc_query_geodata()])
#' @param ... Logical predicates with which to filter the results. Multiple
#' conditions are combined with `&`. Only rows where the condition evaluates to
#' `TRUE` are kept. Accepts normal R expressions as well as any of the special
#' [CQL geometry functions][cql_geom_predicates] such as `WITHIN()` or `INTERSECTS()`.
#' If you know `CQL` and want to write a `CQL` query directly, write it enclosed
#' in quotes, wrapped in the [CQL()] function. e.g., `CQL("ID = '42'")`.
#'
#' If your filter expression contains calls that need to be executed locally, wrap them
#' in `local()` to force evaluation in R before the request is sent to the server.
#'
#' @describeIn filter filter.bcdc_promise
#' @examples
#' \donttest{
#' try(
#'   crd <- bcdc_query_geodata("regional-districts-legally-defined-administrative-areas-of-bc") %>%
#'     filter(ADMIN_AREA_NAME == "Cariboo Regional District") %>%
#'     collect()
#' )
#'
#' try(
#'   ret1 <- bcdc_query_geodata("fire-perimeters-historical") %>%
#'     filter(FIRE_YEAR == 2000, FIRE_CAUSE == "Person", INTERSECTS(crd)) %>%
#'     collect()
#' )
#'
#' # Use local() to force parts of your call to be evaluated in R:
#' try({
#'   # Create a bounding box around two points and use that to filter
#'   # the remote data set
#'   library(sf)
#'   two_points <- st_sfc(st_point(c(1164434, 368738)),
#'                      st_point(c(1203023, 412959)),
#'                      crs = 3005)
#'
#'   # Wrapping the call to `st_bbox()` in `local()` ensures that it
#'   # is executed in R to make a bounding box that is then sent to
#'   # the server for the filtering operation:
#'   res <- bcdc_query_geodata("local-and-regional-greenspaces") %>%
#'     filter(BBOX(local(st_bbox(two_points, crs = st_crs(two_points))))) %>%
#'     collect()
#' })
#' }
#' @export
filter.bcdc_promise <- function(.data, ...) {

  current_cql = cql_translate(..., .colnames = .data$cols_df$col_name %||% character(0))
  ## Change CQL query on the fly if geom is not GEOMETRY
  current_cql = specify_geom_name(.data$cols_df, current_cql)

  # Add cql filter statement to any existing cql filter statements.
  # ensure .data$query_list$CQL_FILTER is class sql even if NULL, so
  # dispatches on sql class and dbplyr::c.sql method is used
  .data$query_list$CQL_FILTER <- c(dbplyr::sql(.data$query_list$CQL_FILTER),
                                   current_cql,
                                   drop_null = TRUE)

  as.bcdc_promise(list(query_list = .data$query_list, cli = .data$cli,
                       record = .data$record, cols_df = .data$cols_df))
}

#' Select columns from bcdc_query_geodata() call
#'
#' Similar to a `dplyr::select` call, this allows you to select which columns you want the Web Feature Service to return.
#' A key difference between `dplyr::select` and `bcdata::select` is the presence of "sticky" columns that are
#' returned regardless of what columns are selected. If any of these "sticky" columns are selected
#' only "sticky" columns are return. `bcdc_describe_feature` is one way to tell if columns are sticky in advance
#' of issuing the Web Feature Service call.
#'
#' @param .data object of class `bcdc_promise` (likely passed from [bcdc_query_geodata()])
#' @param ... One or more unquoted expressions separated by commas. See details.
#'
#' @describeIn select select.bcdc_promise
#'
#' @examples
#' \donttest{
#' try(
#'   feature_spec <- bcdc_describe_feature("bc-airports")
#' )
#'
#' try(
#'   ## Columns that can selected:
#'   feature_spec[feature_spec$sticky == TRUE,]
#' )
#'
#' ## Select columns
#' try(
#'   res <- bcdc_query_geodata("bc-airports") %>%
#'     select(DESCRIPTION, PHYSICAL_ADDRESS)
#' )
#'
#' ## Select "sticky" columns
#' try(
#'   res <- bcdc_query_geodata("bc-airports") %>%
#'     select(LOCALITY)
#' )
#' }
#'
#'
#'@export
select.bcdc_promise <- function(.data, ...){

  ## Eventually have to migrate to tidyselect::eval_select
  ## https://community.rstudio.com/t/evaluating-using-rlang-when-supplying-a-vector/44693/10
  cols_to_select <- tidyselect::vars_select(.data$cols_df$col_name, ...)

  ## id is always added in. web request doesn't like asking for it twice
  cols_to_select <- remove_id_col(cols_to_select)
  ## Always add back in the geom
  cols_to_select <- paste(geom_col_name(.data$cols_df), paste0(cols_to_select, collapse = ","), sep = ",")

  query_list <- c(.data$query_list, propertyName = cols_to_select)

  as.bcdc_promise(list(query_list = query_list, cli = .data$cli,
                       record = .data$record, cols_df = .data$cols_df))

}

#' @importFrom utils head
#' @export
head.bcdc_promise <- function(x, n = 6L, ...) {
  sorting_col <- pagination_sort_col(x$cols_df)
  x$query_list <- c(
    x$query_list,
    count = n,
    sortBy = sorting_col
  )
  x
}

#' @importFrom utils tail
#' @export
tail.bcdc_promise <- function(x, n = 6L, ...) {
  number_of_records <- bcdc_number_wfs_records(x$query_list, x$cli)
  sorting_col <- pagination_sort_col(x$cols_df)
  x$query_list <- c(
    x$query_list,
    count = n,
    sortBy = sorting_col,
    startIndex = number_of_records - n
  )
  x
}


#' @export
names.bcdc_promise <- function(x) {
  cols <- x[["cols_df"]]
  query <- x[["query_list"]]

  if (!is.null(query$propertyName)) {
    select_cols <- strsplit(query$propertyName, ",")[[1]]
    cols <- cols$col_name[cols$sticky | cols$col_name %in% select_cols]
  } else {
    cols <- cols$col_name
  }

  cols[cols == "SHAPE" | cols == "GEOMETRY"] <- "geometry"
  geom_idx <- which(cols == "geometry")

  cols[c(seq_along(cols)[-geom_idx], geom_idx)]

}


#' Throw an informative error when attempting mutate on a `bcdc_promise` object
#'
#' The CQL syntax to generate WFS calls does not current allow arithmetic operations. Therefore
#' this function exists solely to generate an informative error that suggests an alternative
#' approach to use mutate with bcdata
#'
#' @param .data object of class `bcdc_promise` (likely passed from [bcdc_query_geodata()])
#' @param ... One or more unquoted expressions separated by commas. See details.
#' @describeIn mutate mutate.bcdc_promise
#' @examples
#' \donttest{
#'
#' ## Mutate columns
#' try(
#'   res <- bcdc_query_geodata("bc-airports") %>%
#'     mutate(LATITUDE * 100)
#' )
#' }
#'
#'@export
mutate.bcdc_promise <- function(.data, ...){
  dots <- rlang::exprs(...)

  stop(glue::glue(
    "You must type collect() before using mutate() on a WFS. \nAfter using collect() add this mutate call::
    mutate({dots}) "), call. = FALSE)
}


#' Force collection of Web Feature Service request from B.C. Data Catalogue
#'
#' After tuning a query, `collect()` is used to actually bring the data into memory.
#' This will retrieve an sf object into R. The `as_tibble()` function can be used
#' interchangeably with `collect` which matches `dbplyr` behaviour.
#'
#' @param x object of class `bcdc_promise`
#' @inheritParams collect
#' @rdname collect-methods
#' @export
#'
#' @examples
#' \donttest{
#' try(
#'   bcdc_query_geodata("bc-airports") %>%
#'     collect()
#' )
#'
#' try(
#'   bcdc_query_geodata("bc-airports") %>%
#'     as_tibble()
#' )
#' }
#'
collect.bcdc_promise <- function(x, ...){

  x$query_list$CQL_FILTER <- finalize_cql(x$query_list$CQL_FILTER)

  query_list <- x$query_list
  cli <- x$cli

  ## Determine total number of records for pagination purposes
  number_of_records <- bcdc_number_wfs_records(query_list, cli)
  chunk_size <- check_chunk_limit()

  if (number_of_records <= chunk_size) {
    cc <- tryCatch(cli$post(body = query_list, encode = "form"),
                   error = function(e) {
                     stop("There was an issue processing this request.
                     Try reducing the size of the object you are trying to retrieve.", call. = FALSE)})

    catch_wfs_error(cc)
    url <- cc$url
    full_url <- cli$url_fetch(query = query_list)
  } else {
    message(glue::glue("This object has {number_of_records} records and requires {ceiling(number_of_records/chunk_size)} paginated requests to complete."))
    sorting_col <- pagination_sort_col(x$cols_df)

    query_list <- c(query_list, sortby = sorting_col)

    # Create pagination client
    cc <- crul::Paginator$new(
      client = cli,
      by = "limit_offset",
      limit_param = "count",
      offset_param = "startIndex",
      limit = number_of_records,
      chunk = chunk_size,
      progress = interactive()
    )

    message("Retrieving data")

    tryCatch(cc$post(body = query_list, encode = "form"),
             error = function(e) {
               stop("There was an issue processing this request.
                     Try reducing the size of the object you are trying to retrieve.", call. = FALSE)})

    url <- cc$url
    full_url <- cc$url_fetch(query = query_list)

    catch_wfs_error(cc)
  }

  txt <- cc$parse("UTF-8")

  as.bcdc_sf(bcdc_read_sf(txt), query_list = query_list, url = url,
             full_url = full_url)
}


#' @inheritParams collect.bcdc_promise
#' @rdname collect-methods
#' @export
as_tibble.bcdc_promise <- collect.bcdc_promise

#' Show SQL and URL used for Web Feature Service request from B.C. Data Catalogue
#'
#' Display Web Feature Service query CQL
#'
#' @param x object of class bcdc_promise or bcdc_sf
#' @inheritParams show_query
#' @describeIn show_query show_query.bcdc_promise
#'
#' @export
#' @examples
#' \donttest{
#' try(
#'   bcdc_query_geodata("bc-environmental-monitoring-locations") %>%
#'     filter(PERMIT_RELATIONSHIP == "DISCHARGE") %>%
#'     show_query()
#' )
#'   }
#'
show_query.bcdc_promise <- function(x, ...) {

  y <- list()
  y$base_url <- x$cli$url
  y$query_list <- x$query_list
  y$query_list$CQL_FILTER <- finalize_cql(y$query_list$CQL_FILTER)
  y$full_url <- x$cli$url_fetch(query = y$query_list)

  as.bcdc_query(y)

}



#' @describeIn show_query show_query.bcdc_promise
#'
#' @export
#' @examples
#' \donttest{
#' try(
#'   air <- bcdc_query_geodata("bc-airports") %>%
#'     collect()
#' )
#'
#' try(
#'   show_query(air)
#' )
#' }
show_query.bcdc_sf <- function(x, ...) {

  y <- list()
  y$url <- attr(x, "url")
  y$query_list <- attr(x, "query_list")
  y$full_url <- attr(x, "full_url")

  as.bcdc_query(y)
}

# collapse vector of cql statements into one
finalize_cql <- function(x, con = wfs_con) {
  if (is.null(x) || !length(x)) return(NULL)
  dbplyr::sql_vector(x, collapse = " AND ", con = con)
}

cat_line_wrap <- function(..., indent = 0, exdent = 1, col = NULL, background_col = NULL, file = stdout()) {
  txt <- strwrap(paste0(..., collapse = ""), indent = indent, exdent = exdent)
  cat_line(txt, col = col, background_col = background_col, file = file)
}

#' @export
"[.bcdc_recordlist" <- function(x, i, j, ..., drop = FALSE) {
  out <- unclass(x)[i]
  as.bcdc_recordlist(out)
}