R/thingsboard_api.R

In Rthingsboard: 'ThingsBoard' API

#' @title Thingboard API Class
#'
#' @field url [character] URL of the 'ThingsBoard' IoT platform.
#' @field publicId [character] the public ID of the device
#' @field token [character] the current token
#' @field tokenTimeOut A [numeric] contains the time out of a token in seconds (default 300)
#' @field tokenEpiration A [numeric] with the Epoch of the expiration date time of current token
#'
#' @export ThingsboardApi
#' @exportClass ThingsboardApi
#' @importFrom methods new
#'
#' @seealso The `ThingsboardApi` class methods :
#' - [ThingsboardApi_checkToken] for checking and refreshing the token
#' - [ThingsboardApi_getToken] for getting authorisation token from Thingsboard server for a specific device
#' - [ThingsboardApi_getKeys] for fetching data keys of an entity
#' - [ThingsboardApi_getValues] and [ThingsboardApi_getTelemetry] for fetching telemetry of an entity
#'
#' @examples
#' \donttest{
#' thinksboard_api = tryCatch(
#'   {
#'     ThingsboardApi(url="http://scada.g-eau.fr",
#'                    publicId="299cedc0-f3e9-11e8-9dbf-cbc1e37c11e3")
#'   },
#'   error = function(e) {
#'     message("An error occured:\n", e)
#'     return(FALSE)
#'   }
#' )
#' }
ThingsboardApi <- setRefClass(
  "ThingsboardApi",

  fields = list(
    url = "character",
    publicId = "character",
    token = "character",
    tokenExpiration = "numeric",
    tokenTimeOut = "numeric"
  ),

  methods = list(
    initialize = function(..., tokenTimeOut = 300) {
      callSuper(..., tokenTimeOut = tokenTimeOut)
      getToken()
    }
  )
)


#' Check if the token is timeouted and refresh it if necessary
#'
#' @description
#' This method is automatically call by each other methods of the class [ThingsboardApi].
#' So, except for debugging purpose, it's not useful to call directly.
#'
#' @name ThingsboardApi_checkToken
#' @return [NULL]
#'
NULL
ThingsboardApi$methods(
  checkToken = function () {
    if (as.numeric(Sys.time()) >= tokenExpiration) {
      getToken()
    }
  }
)


#' Get authorisation token from thingsboard server for a specific device
#'
#' @description
#' This method is automatically called by [ThingsboardApi_checkToken], and so by any other methods
#' of the class [ThingsboardApi] as needed. Except for debugging purpose, it's not useful to call directly.
#'
#' @name ThingsboardApi_getToken
#' @param timeOut [numeric] number of second before token timeout (default field `tokenTimeOut`)
#'
#' @return A [list] with keys 'token' and 'refreshtoken'
#'
NULL
ThingsboardApi$methods(
  getToken = function (timeOut = tokenTimeOut) {
    res <- httr::POST(
      url = file.path(url, "api/auth/login/public"),
      body = list(publicId = publicId),
      encode = "json"
    )
    if (httr::http_error(res) || httr::http_status(res)$reason != "OK") {
      stop("Request failed with status ",
           httr::http_status(res)$message)
    }

    dToken = httr::content(res, as = "parsed", encoding = "Latin1")
    .self$token <- dToken$token
    logger::log_debug("ThingsboardApi$getToken: ", substr(token, 1, 12), "...")
    .self$tokenExpiration <- as.numeric(Sys.time()) + timeOut
    logger::log_debug("ThingsboardApi$getToken: expiration ",
                      as.character(as.POSIXct(tokenExpiration, origin =
                                                "1970-01-01")))
    return (dToken)
  }
)


#' Fetch data keys for an entity
#'
#' @name ThingsboardApi_getKeys
#'
#' @details
#' The description of this operation in API documentation is here: <https://thingsboard.io/docs/user-guide/telemetry/#get-telemetry-keys>
#'
#' @param entityId [character] entity ID
#' @param entityType [character] (default "DEVICE")
#'
#' @return A vector of [character] with the keys available for the requested device.
#'
NULL
ThingsboardApi$methods(
  getKeys = function(entityId, entityType = "DEVICE") {
    checkToken()
    res = httr::GET(
      url = file.path(
        url,
        "api/plugins/telemetry",
        entityType,
        entityId,
        "keys/timeseries"
      ),
      httr::content_type_json(),
      httr::add_headers(`X-Authorization` = paste("Bearer", token))
    )

    if (httr::http_error(res) || httr::http_status(res)$reason != "OK") {
      stop("Request failed with status ",
           httr::http_status(res)$message)
    }

    keys <-
      unlist(httr::content(res, as = "parsed", encoding = "Latin1"))
    logger::log_debug(paste("keys =", paste(keys, collapse = ", ")))
    return (keys)
  }
)


#' Fetch telemetry
#'
#' @description
#' See [ThingsboardApi_getTelemetry].
#'
#' @name ThingsboardApi_getValues
#'
NULL
ThingsboardApi$methods(
  getValues = function(entityId,
                       keys,
                       startTs,
                       endTs,
                       interval = NULL,
                       agg = "NONE",
                       entityType = "DEVICE") {
    checkToken()
    lQuery = list(
      keys = paste(lapply(keys, curl::curl_escape), collapse = ","),
      startTs = Date2EpochMilli(startTs),
      endTs = Date2EpochMilli(endTs),
      agg = agg
    )
    if (!is.null(interval)) {
      query$interval <- interval
    }
    query <- paste(sapply(names(lQuery),
                          function(x) {
                            paste0(x, "=", lQuery[[x]])
                          }), collapse = "&")

    logger::log_debug("getValues query ", query)

    res = httr::GET(
      url = file.path(
        url,
        "api/plugins/telemetry",
        entityType,
        entityId,
        "values/timeseries"
      ),
      query = query,
      httr::content_type_json(),
      httr::add_headers(`X-Authorization` = paste("Bearer", token))
    )

    if (httr::http_error(res) || httr::http_status(res)$reason != "OK") {
      stop("Request failed with status ",
           httr::http_status(res)$message)
    }

    lV <- httr::content(res, as = "parsed", encoding = "Latin1")
    lV <- lapply(names(lV),
                 function(x) {
                   df <- data.frame(key = x,
                                    v = matrix(unlist(lV[[x]]), ncol = 2, byrow = TRUE),
                                    stringsAsFactors = FALSE)
                   colnames(df) <-
                     c("key", "ts", "value")
                   df
                 })
    if(length(lV) > 0) {
      dfV <- do.call(rbind, lV)
      dfV$ts <- EpochMilli2Date(dfV$ts, timezone = attributes(startTs)$tzone)
      dfV$value <- as.numeric(dfV$value)
    } else {
      dfV <- data.frame(key = character(),
                        ts = as.POSIXct(character()),
                        value = numeric(),
                        stringsAsFactors = FALSE)
    }
    return(dfV)
  }
)


#' Fetch telemetry
#'
#' @description
#' Fetch telemetry data of an entity.
#'
#' It uses the following API: <https://thingsboard.io/docs/user-guide/telemetry/#get-telemetry-values>
#'
#' The method `getValues` has a strong limitation as the 'ThingsBoard' API only send the
#' 100 last values of each key. The method `getTelemetry` overcomes this limitation by
#' automatically by calling `getValues` in a loop.
#'
#'
#' @name ThingsboardApi_getTelemetry
#' @rdname ThingsboardApi_getTelemetry
#'
#' @param entityId A [character] with the entity ID given (See <https://thingsboard.io/docs/user-guide/entity-views/>)
#' @param keys Vector of [character] with the list of keys from which getting the telemetry values
#' @param entityType A [character] (default "DEVICE")
#' @param startTs A [numeric] or a [POSIXct] representing respectively the epoch or the date of the start of data extraction period
#' @param endTs A [numeric] or a [POSIXct] representing respectively the epoch or the date of the end of data extraction period
#'
#' @return A [data.frame] with one row per data and 3 columns:
#'
#'  * `key`: a [character] with the key
#'  * `ts`: a [POSIXct] with the timestamp of the data
#'  * `value`: a [numeric] with the value of the data
#'
#' @import dplyr
#'
NULL
ThingsboardApi$methods(
  getTelemetry = function(..., endTs) {
    l <- list()
    i <- 0
    while (TRUE) {
      i <- i + 1
      l[[i]] <- getValues(..., endTs = endTs)
      if (nrow(l[[i]]) == 0) break
      df_minTs <- l[[i]] %>% arrange(ts) %>% group_by(key) %>% slice(1)
      endTs <- max(df_minTs$ts) - 1
    }
    df <- do.call(rbind, l)
    df <- unique(df)
    return(df)
  }
)