Info | ||
---|---|---|
| ||
This page is originally from a git repository where its content can be managed and presented in a more dynamical way. This is a snapshot as of today ( ). There will be no frequent updates or alike. New content comes with a new documentation platform – some day! |
Table of Contents | ||
---|---|---|
|
input:
Per API: Search for all items of contact x, i) by (system) email address, ii) by contact names.
/sensors/device/getItemsOfContact/{contactEmail}
OR/sensors/contacts/getContactsByName/{firstName}/{lastName}
--> email address will be output, then go on with step 1input:
/sensors/device/getItemsOfContact/{contactEmail}
OR/sensors/contacts/getContactsByName/{firstName}/{lastName}
--> email address as output then go to step 1/sensors/contacts/getAllContactRoles
--> pick out which ID is related to which role type (e.g. Data scientist="id": 29]Code Block | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|
| ||||||||||
''' basic documentation of sensor.awi.de REST-API can be found here: https://sensor.awi.de/api/ ''' ## necessary libraries import json import requests import datetime ## PROD #sensorURL<- "https://sensor.awi.de/rest/sensors/" ## SANDBOX sensorURL <- "https://sandbox.sensor.awi.de/rest/sensors/" ## + ================================ | xxx | ================================ + ## ''' USE CASE 1: We are looking for a list of items that have a certain user in their contact list. 1.1.a We know the contact email address (as in sensor.awi.de) of the user. ''' ## + ================================ | xxx | ================================ + ## email = 'marcel.nicolaus@awi.de' ## create link for API call link = sensorURL + "device/getItemsOfContact/" + email ## calling api answer = requests.get(link) ## the content itself, read as a json object fullList = json.loads(answer.content) ## status of request (see httpstatuses.com/) answer.status_code ## since the result can be 0 items, 1 item, and >1 items , we filter the item ids according the number of items and write to a list if len(fullList) == 0: itemList = [] elif len(fullList) == 1: itemList = [fullList[0]['id']] else: itemList = [i['id'] for i in fullList] ## there we go: print(itemList) ## + ================================ | xxx | ================================ + ## ''' USE CASE 1: We are looking for a list of items that have a certain user in their contact list. 1.1.b We know the last name (and if necessary the first name) of the user. Aim: get a single email address. ''' ## + ================================ | xxx | ================================ + ## ## creation of the API call link = sensorURL + 'contacts/getAllSystemContacts' ## CASE: unique last name or no contact found ## variables lastName = 'Gerchow' ## calling api answer = requests.get(link) x = json.loads(answer.content) ## loop over entire list for a in x: if a['lastName'] == lastName: print(a['email']) ## CASE: non-unique last name or no contact ## variables lastName = 'Nicolaus' firstName = 'Rita' ## calling api answer = requests.get(link) x = json.loads(answer.content) ## loop over entire list for a in x: if a['lastName'] == lastName and a['firstName'] == firstName: print(a['email']) ## + ================================ | xxx | ================================ + ## ''' USE CASE 1: We are looking for a list of items that have a certain user in their contact list. 1.2. We have info from 1.1 but need that user only with a certain contact role (e.g. 'Principal Investigator'). ''' ## + ================================ | xxx | ================================ + ## ## which user should be searched? ## remember 'email' from 1.1? #email = 'sandra.tippenhauer@awi.de' ## what role should be filtered? inputRole = 'Principal Investigator' inputRole = 'Editor' ## create empty lists lstID, lstURN = [], [] ## recycling of itemList from 1.1 for item in itemList: link = sensorURL + "contacts/getDeviceContacts/" + str(item) answer = requests.get(link) x = json.loads(answer.content) ## create dictionaries to map uuids to vocables dictRoles = {} dictAccounts = {} for i in x: if isinstance(i['contact'], dict): dictAccounts[i['contact']['@uuid']] = i['contact']['email'] if isinstance(i['vocable'], dict): dictRoles[i['vocable']['@uuid']] = i['vocable']['generalName'] ## extracting from item user, role = [], [] for i in x: if isinstance(i['vocable'], dict): role.append(i['vocable']['@uuid']) elif isinstance(i['vocable'], str): role.append(i['vocable']) else: pass ## if isinstance(i['contact'], dict): user.append(i['contact']['@uuid']) elif isinstance(i['contact'], str): user.append(i['contact']) else: pass ## merging info from item and mapping dict for i in range(len(user)): ## <-- could improve!!! if dictRoles[role[i]] == inputRole and dictAccounts[user[i]] == email: lstID.append(item) lstURN.append(x[0]['item']['urn']) ## neat output as some sort of report ... print(json.dumps({'user':email, 'timeOfQuery':datetime.datetime.isoformat(datetime.datetime.utcnow()), 'role':inputRole, 'urn':lstURN, 'id':lstID}) ) ## ... or just as a single list if item IDs print(lstID) ## eof |
Code Block | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|
| ||||||||||
## ''' ## basic documentation of sensor.awi.de REST-API can be found here: https://sensor.awi.de/api/ ## ''' settingUpThings <- function(x){ mirror <- 'https://ftp.gwdg.de/pub/misc/cran' newPackages <- x[!(x %in% installed.packages()[,'Package'])] if(length(newPackages)){ install.packages(newPackages, dep = TRUE, repos = mirror) } lapply(x, library, character.only = TRUE, logical.return = TRUE) } libsToBeLoaded <- c('jsonlite', 'plyr', 'lubridate', 'httr') settingUpThings(libsToBeLoaded) ## PROD #sensorURL<- "https://sensor.awi.de/rest/sensors/" ## SANDBOX sensorURL <- "https://sandbox.sensor.awi.de/rest/sensors/" ## + ================================ | xxx | ================================ + ## ## ''' ## USE CASE 1: We are looking for a list of items that have a certain user in their contact list. ## 1.1.a We know the contact email address (as in sensor.awi.de) of the user. ## ''' ## + ================================ | xxx | ================================ + ## email <- 'marcel.nicolaus@awi.de' ## create link for API call link <- paste0(sensorURL, "device/getItemsOfContact/", email) ## calling api answer <- GET(link) ## the content itself, read as a json object fullList <- content(answer, as = 'parsed') ## status of request (see httpstatuses.com/) answer$status_code ## since the result can be 0 items, 1 item, and >1 items , we filter the item ids according the number of items and write to a list if (length(fullList) == 0){ itemList <- list() } else if (length(fullList) == 1){ itemList <- list(fullList[[1]]$id) } else { itemList <- list() for (i in seq_along(fullList)){ itemList[[i]] <- fullList[[i]]$id } } ## there we go: print(itemList) ## + ================================ | xxx | ================================ + ## ## ''' ## USE CASE 1: We are looking for a list of items that have a certain user in their contact list. ## 1.1.b We know the last name (and if necessary the first name) of the user. Aim: get a single email address. ## ''' ## + ================================ | xxx | ================================ + ## ## creation of the API call link <- paste0(sensorURL,'contacts/getAllSystemContacts') ## CASE: unique last name or no contact found ## variables lastName <- 'Gerchow' ## calling api answer <- GET(link) x <- content(answer, as = 'parsed') for (i in seq_along(x)){ if ( is.null(x[[i]]$lastName) == FALSE && x[[i]]$lastName == lastName ) { print(x[[i]]$email) } } ## CASE: non-unique last name or no contact ## variables lastName <- 'Nicolaus' firstName <- 'Rita' ## calling api answer <- GET(link) x <- content(answer, as = 'parsed') ## loop over entire list for (i in seq_along(x)){ if ( is.null(x[[i]]$lastName) == FALSE && x[[i]]$lastName == lastName && x[[i]]$firstName == firstName ) { print(x[[i]]$email) } } ## + ================================ | xxx | ================================ + ## ## ''' ## USE CASE 1: We are looking for a list of items that have a certain user in their contact list. ## 1.2. We have info from 1.1 but need that user only with a certain contact role (e.g. 'Principal Investigator'). ## ''' ## + ================================ | xxx | ================================ + ## ## which user should be searched? ## remember 'email' from 1.1? #email = 'sandra.tippenhauer@awi.de' ## what role should be filtered? inputRole <- 'Principal Investigator' inputRole <- 'Editor' lstID <- NULL lstURN <- NULL for (item in seq_along(itemList)){ link <- paste0(sensorURL, "contacts/getDeviceContacts/", itemList[[item]]) print(link) answer <- GET(link) x <- content(answer, as = 'parsed') ## create dictionaries to map uuids to vocables dictRoles <- NULL dictAccounts <- NULL for (i in seq_along(x)){ ## print(x[[i]]) if (is.list(x[[i]]$contact) == TRUE){ dictAccounts[x[[i]]$contact$'@uuid'] <- x[[i]]$contact$email } if (is.list(x[[i]]$vocable) == TRUE){ dictRoles[x[[i]]$vocable$'@uuid'] <- x[[i]]$vocable$generalName } } ## extracting from item user <- NULL role <- NULL for (i in seq_along(x)){ if (is.list(x[[i]]$vocable) == TRUE){ role <- append(role, x[[i]]$vocable$'@uuid') } else { role <- append(role, x[[i]]$vocable) } ## if (is.list(x[[i]]$contact) == TRUE){ user <- append(user, x[[i]]$contact$'@uuid') } else { user <- append(user, x[[i]]$contact) } } ## merging info from item and mapping dict for (i in seq_along(x)){ if ( dictRoles[role[i]] == inputRole && dictAccounts[user[i]] == email){ lstID <- append(lstID, item) lstURN <- append(lstURN, x[[1]]$item$urn) } } } ## neat output as some sort of report ... print( toJSON(list(user = email, timeOfQuery = now(tz = "GMT"), role = inputRole, urn = lstURN, id = lstID ) ) ) ## ... or just as a single list if item IDs print(lstID) ## eof |
Precondition: only valid for current configuration. If the item was object to modifications (mounts) the results may differ.
input:
https://sensor.awi.de/rest/sensors/item/getChildrenOfItem/{itemID}
https://sensor.awi.de/rest/sensors/sensorOutputs/getAllSensorOutputTypes
-> filter by i.e. regex "temp" (water temperature, technical temperature, etc. ...)https://sensor.awi.de/rest/sensors/sensorOutputs/getDeviceSensorOutputs/{itemID}
TODO: solve outputType is matching exactly but the response is not unique.... doh!
Code Block | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|
| ||||||||||
'''
basic documentation of sensor.awi.de REST-API can be found here: https://sensor.awi.de/api/
'''
## necessary libraries
import json
import requests
import re
import itertools
## PROD
#sensorURL = "https://sensor.awi.de/rest/sensors/"
## SANDBOX
sensorURL = "https://sandbox.sensor.awi.de/rest/sensors/"
## + ================================ | xxx | ================================ + ##
'''
USE CASE 3: give all items that measure parameter xyz on platform zyx
'''
## + ================================ | xxx | ================================ + ##
## -------------------------------- 1 -------------------------------- ##
urn = 'vessel:polarstern'
#urn = 'vessel:uthoern'
#urn = 'aircraft:polar7'
link = sensorURL + '/item/getItemByUrn/' + urn
answer = requests.get(link)
## check
answer.status_code
item = json.loads(answer.content)
## aux function, saves code later
def getKids(ID):
link = sensorURL + 'item/getChildrenOfItem/' + str(ID)
answer = requests.get(link)
x = json.loads(answer.content)
y = [f['id'] for f in x]
return(y)
## there might be smarter ways ...
idCollection = [item['id']]
initCount = 0
afterCount = len(idCollection)
while initCount != afterCount:
## print(idCollection)
## retrieve kids
a = [getKids(ii) for ii in idCollection]
## flatten the list and remove '[]'
subs = list(itertools.chain(*a))
## set first counter
initCount = len(idCollection)
## combine lists
idCollection = idCollection + subs
## kickout duplicates
idCollection = list(dict.fromkeys(idCollection))
## set second counter
afterCount = len(idCollection)
## -------------------------------- 2 -------------------------------- ##
## should be unique
## see 'thermal diffusivity' vs. 'thermosalinograph'
pattern = 'temperature'
link = sensorURL + "vocabulary/getAllVocables?pointInTime=2015-01-01T00%3A00%3A00.000Z"
answer = requests.get(link)
x = json.loads(answer.content)
param = []
for i in x:
## first occurence gives uuid
if isinstance(i['vocableGroup'], dict) and i['vocableGroup']['name'] == 'SensorOutputTypes':
vocGroup = i['vocableGroup']['@uuid']
## increase chance for success by case insensitive matching
if str(pattern).lower() in str(i['systemName']).lower():
## guarantee 1st occurence in list
if isinstance(i['vocableGroup'], dict) and i['vocableGroup']['@uuid'] == vocGroup:
param.append(i)
## guarantee all other occurences in list
if isinstance(i['vocableGroup'], str) and i['vocableGroup'] == vocGroup:
param.append(i)
## if not unique ( len(param) == 1) stop here
if len(param) != 1:
print('result must be unique (as in a single result)')
print('')
raise Exception('break')
else:
print(param)
## -------------------------------- 3 -------------------------------- ##
param = [param[0]]
parameterID = []
## remember me?
for i in idCollection:
print(i)
link = sensorURL + "sensorOutputs/getDeviceSensorOutputs/" + str(i)
answer = requests.get(link)
x = json.loads(answer.content)
## creating dict to map values
dic = {}
## just to create dict
for output in x:
if isinstance(output['sensorOutputType'], dict):
dic[output['sensorOutputType']['systemName']] = output['sensorOutputType']['@uuid']
## checking if par exists
if param[0]['systemName'] not in dic:
pass
else:
## true run
for output in x:
## 1st ofccurence
if isinstance(output['sensorOutputType'], dict) and output['sensorOutputType']['@uuid'] == dic[param[0]['systemName']]:
parameterID.append(output['id'])
## all other occurences
elif isinstance(output['sensorOutputType'], str) and output['sensorOutputType'] == dic[param[0]['systemName']]:
parameterID.append(output['id'])
## something weird going on
else:
pass
if len(parameterID) != 0:
print('found something')
## optional:
## -------------------------------- 4 -------------------------------- ##
for jj in parameterID:
link = sensorURL + 'sensorOutputs/getSensorOutput/' + str(jj)
answer = requests.get(link)
x = json.loads(answer.content)
print(x)
## eof
|
Code Block | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|
| ||||||||||
## '''
## basic documentation of sensor.awi.de REST-API can be found here: https://sensor.awi.de/api/
## '''
settingUpThings <- function(x){
mirror <- 'https://ftp.gwdg.de/pub/misc/cran'
newPackages <- x[!(x %in% installed.packages()[,'Package'])]
if(length(newPackages)){
install.packages(newPackages, dep = TRUE, repos = mirror)
}
lapply(x, library, character.only = TRUE, logical.return = TRUE)
}
libsToBeLoaded <- c('jsonlite', 'plyr', 'stringr', 'httr')
settingUpThings(libsToBeLoaded)
## PROD
#sensorURL<- "https://sensor.awi.de/rest/sensors/"
## SANDBOX
sensorURL <- "https://sandbox.sensor.awi.de/rest/sensors/"
## + ================================ | xxx | ================================ + ##
## '''
## USE CASE 3: give all items that measure parameter xyz on platform zyx
## '''
## + ================================ | xxx | ================================ + ##
## -------------------------------- 1 -------------------------------- ##
urn <- 'vessel:polarstern'
#urn <- 'vessel:uthoern'
#urn <- 'aircraft:polar7'
link <- paste0(sensorURL, '/item/getItemByUrn/', urn)
answer <- GET(link)
## check
answer$status_code
item <- content(answer, as = 'parsed')
getKids <- function(ID){
link <- paste0(sensorURL, 'item/getChildrenOfItem/', ID)
answer <- GET(link)
x <- content(answer, as = 'parsed')
y <- NULL
for (f in seq_along(x)){
y[f] <- x[[f]]$id
}
return(y)
}
## there might be smarter ways ...
idCollection <- c(item$id)
initCount <- 0
afterCount <- length(idCollection)
while (initCount != afterCount){
print(idCollection)
## set first counter
initCount <- length(idCollection)
## retrieve kids and combine lists
for (ii in seq_along(idCollection)){
idCollection <- c(idCollection, getKids(idCollection[ii]))
}
## skipping some foo, not necessary here
## ...
## kickout duplicates
idCollection <- unique(idCollection)
## set 2nd counter
afterCount <- length(idCollection)
}
## -------------------------------- 2 -------------------------------- ##
## should be unique
## see 'thermal diffusivity' vs. 'thermosalinograph'
pattern = 'tempe'
link <- paste0(sensorURL, "vocabulary/getAllVocables?pointInTime=2015-01-01T00%3A00%3A00.000Z")
answer <- GET(link)
x <- content(answer, as = 'parsed')
param <- list()
for (i in seq_along(x)){
## 1st occurence gives uuid
if ( is.list(x[[i]]$vocableGroup) == TRUE && x[[i]]$vocableGroup$name == 'SensorOutputTypes' ){
vocGroup <- x[[i]]$vocableGroup$'@uuid'
}
## increase chance for success by case insensitive matching
if ( str_detect(tolower(x[[i]]$systemName), tolower(pattern)) ){
## guarantee 1st occurence in list
if ( is.list(x[[i]]$vocableGroup) == TRUE && x[[i]]$vocableGroup$'@uuid' == vocGroup ){
param[[length(param)+1]] <- x[[i]]
}
## guarantee all other occurences in list
if( is.character(x[[i]]$vocableGroup) == TRUE && x[[i]]$vocableGroup == vocGroup ){
param[[length(param)+1]] <- x[[i]]
}
}
}
if ( length(param) != 1 ){
print('result must be unique (as in a single result)')
} else {
print(param)
}
## -------------------------------- 3 -------------------------------- ##
param <- param[[1]] ## only temp, only for now
parameterID <- NULL
## remember me?
for (i in seq_along(idCollection)){
link <- paste0(sensorURL, "sensorOutputs/getDeviceSensorOutputs/", 61)# idCollection[i])
link <- paste0(sensorURL, "sensorOutputs/getDeviceSensorOutputs/", idCollection[i])
answer <- GET(link)
x <- content(answer, as = 'parsed')
## creating dict to map values
dic <- list()
## just to create dict
for (output in seq_along(x)){
## print(i)
if ( is.list(x[[output]]$sensorOutputType) == TRUE ){
dic[x[[output]]$sensorOutputType$systemName] <- x[[output]]$sensorOutputType$'@uuid'
}
}
## checking if par exists
if ( param$systemName %in% names(dic) == TRUE ){
## true run
for (output in seq_along(x)){
## 1st occurence
if ( is.list(x[[output]]$sensorOutputType) == TRUE && x[[output]]$sensorOutputType$'@uuid' == dic[param$systemName] ){
parameterID <- append(parameterID, x[output][[1]]$id)
}
## all other occurences
else if ( is.character(x[[output]]$sensorOutputType) == TRUE && x[[output]]$sensorOutputType == dic[param$systemName] ){
parameterID <- append(parameterID, x[output][1]$id)
} else {
next
}
}
}
}
if ( length(parameterID) > 0 ){
print('found something')
}
## optional:
## -------------------------------- 4 -------------------------------- ##
for (jj in seq_along(parameterID)){
link <- paste0(sensorURL, 'sensorOutputs/getSensorOutput/', parameterID[jj])
answer <- GET(link)
x <- content(answer, as = 'parsed')
print(x)
}
## eof
|
input:
https://sensor.awi.de/rest/sensors/contacts/login
https://sensor.awi.de/rest/sensors/contacts/getAllContactRoles
https://sensor.awi.de/rest/sensors/item/getItemByUrn/
https://sensor.awi.de/rest/sensors/contacts/putExistingContactToDevice/{itemID}/{userID}/{contactRoleID}
https://sensor.awi.de/rest/sensors/contacts/deleteContactFromDevice/{itemID}/{userID}/{contactRoleID}
Code Block | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|
| ||||||||||
''' basic documentation of sensor.awi.de REST-API can be found here: https://sensor.awi.de/api/ ''' ## necessary libraries import re import json import requests ## -------------------------------- 1 -------------------------------- ## userName = 'norbert.anselm@awi.de' PW = # <- sensitive information, do not put in git, spaces or elsewhere!!! ## POST request to sensor ## with payload username and password <- this is plain text, keep it as secure as possible!! ## PROD sensorURL = 'https://sensor.awi.de/rest/' ## SANDBOX sensorURL = 'https://sandbox.sensor.awi.de/rest/' auth = requests.post(sensorURL + 'sensors/contacts/login', data = {'username': userName,'authPassword': PW} ) ## check httpstatuses.com/ for translation ## check if the request was successful if auth.status_code == 200: print('we are happy') ## extracting token -> keep it token = auth.cookies['x-auth-token'] else: print('nope, no token available') ## -------------------------------- 2 -------------------------------- ## link = sensorURL + 'sensors/contacts/getAllContactRoles' answer = requests.get(link) allRoles = json.loads(answer.content) ## creating a dict to translate role string representation to numeric representation dictRoles = {} for i in allRoles: dictRoles[i['generalName']] = i['id'] ## -------------------------------- 3 -------------------------------- ## link = sensorURL + 'sensors/item/getItemByUrn/' ## explicitly on sandbox urn = 'anemometer:aaaa' #'vessel:uthoern:moses_moblab:ipu443_awi_01' answer = requests.get(link + urn) item = json.loads(answer.content) if auth.status_code == 200: itemID = item['id'] else: print('no item available') ## -------------------------------- 4 -------------------------------- ## ## ================ 1 we have user ID ================ ## userID = 442 x = json.loads(requests.get(sensorURL + 'sensors/contacts/getContact/' + str(userID)).content) ## checking ... x['email'] ## ================ 2 we know system email ================ ## user = 'jaeilers@awi.de' link = sensorURL + 'sensors/contacts/getContactsByEmail/' x = json.loads(requests.get(link + user).content) if len(x) == 1: userID = x[0]['id'] print(x) elif len(x) == 0: print('no user') else: ## just for the case print('\n choose 1: \n') print(x) ## ================ 3 first and last name ================ ## link = sensorURL + 'sensors/contacts/getContactsByName/' #norbert/anselm firstName = 'norbert' lastName = 'anselm' x = json.loads(requests.get(link + firstName + '/' + lastName).content) if len(x) == 1: userID = x[0]['id'] elif len(x) == 0: print('no user') else: print('\nchoose 1!\n') print(x) ## -------------------------------- 5 -------------------------------- ## ## ================ A -> add certain users ================ ## ## constructing the post link ## input see above! link = sensorURL + 'sensors/contacts/putExistingContactToDevice/' + str(itemID) + '/' + str(userID) + '/' + str(dictRoles['Device Contact']) ## submit the cooky a = requests.put(link, headers = {'content-type':'application/json'}, cookies = {'x-auth-token': token}) if a.status_code != 200: print('we are NOT happy! status code -> ' + str(a.status_code)) else: print('we are happy! status code: ' + str(a.status_code) + ' -> ' + a.reason) ## ================ B -> remove certain users ================ ## ## analogue to put link = sensorURL + 'sensors/contacts/deleteContactFromDevice/' + str(itemID) + '/' + str(userID) + '/' + str(dictRoles['Device Contact']) ## hm, wait for SE... <500> a = requests.delete(link, headers = {'content-type':'application/json'}, cookies = {'x-auth-token': token}) if a.status_code != 200: print('we are NOT happy! status code -> ' + str(a.status_code)) else: print('we are happy! status code: ' + str(a.status_code) + ' -> ' + a.reason) |
Code Block | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|
| ||||||||||
settingUpThings <- function(x){ ## x == tuple of libs ## installing and loading necessary libs mirror <- 'https://ftp.gwdg.de/pub/misc/cran' ## designated VM path ## lib <- '/home/edvs1/nanselm/R/x86_64-pc-linux-gnu-library' newPackages <- x[!(x %in% installed.packages()[,'Package'])] if(length(newPackages)){ install.packages(newPackages, dep = TRUE, repos = mirror)#, lib = lib) } lapply(x, library, character.only = TRUE, logical.return = TRUE) } libsToBeLoaded <- c('jsonlite', 'plyr', 'httr') settingUpThings(libsToBeLoaded) ## PROD sensorURL<- "https://sensor.awi.de/rest/" ## SANDBOX sensorURL <- "https://sandbox.sensor.awi.de/rest/" ## -------------------------------- 1 -------------------------------- ## userName <- 'norbert.anselm@awi.de' ##PW <- ## link <- paste0(sensorURL, 'sensors/contacts/login') a <- POST(url = link , body = list("username" = userName, "authPassword" = PW) , encode = "form" ) if ( a$status_code == 200 ) { print('we are happy') token <- a$cookies$value[2] } else { print('nope, no token ovoiloble') } ## -------------------------------- 2 -------------------------------- ## link <- paste0(sensorURL, 'sensors/contacts/getAllContactRoles') ## JSON way answer <- read_json(link) dictRoles <- list() for (i in seq_along(answer)){ dictRoles[[answer[[i]]$generalName]] <- answer[[i]]$id } ## httr way answer <- GET(link) allRoles <- content(answer, as = 'parsed') dictRoles <- list() for (i in seq_along(allRoles)){ dictRoles[[allRoles[[i]]$generalName]] <- allRoles[[i]]$id } ## -------------------------------- 3 -------------------------------- ## link <- paste0(sensorURL,'sensors/item/getItemByUrn/') ## explicitly on sandbox urn = 'anemometer:aaaa' answer <- GET(paste0(link,urn)) if ( answer$status_code == 200 ) { print('we are happy') item <- content(answer, as = 'parsed') itemID <- item$id } else { print('nope, no item available') } ## -------------------------------- 4 -------------------------------- ## ## ================ 1 we have user ID ================ ## userID <- 442 link <- paste0(sensorURL, 'sensors/contacts/getContact/', userID) answer <- GET(paste0(link)) x <- content(answer, as = 'parsed') ## test x$email ## ================ 2 we know system email ================ ## user <- 'jaeilers@awi.de' link <- paste0(sensorURL, 'sensors/contacts/getContactsByEmail/', user) answer <- GET(paste0(link)) x <- content(answer, as = 'parsed') if ( length(x) == 1 ){ userID <- x[[1]]$id } else if ( length(x) == 0 ){ print('no user') } else { ## <- unlikely print('choose 1!') print(x) } ## ================ 3 first and last name ================ ## firstName <- 'janik' lastName <- 'anselm' link <- paste0(sensorURL, 'sensors/contacts/getContactsByName/', firstName, '/', lastName) answer <- GET(paste0(link)) x <- content(answer, as = 'parsed') if ( length(x) == 1 ){ userID <- x[[1]]$id } else if ( length(x) == 0 ){ print('no user') } else { ## <- unlikely print('choose 1!') print(x) } ## -------------------------------- 5 -------------------------------- ## ## ================ A -> add certain users ================ ## ## constructing the post link ## input see above! link <- paste0(sensorURL, 'sensors/contacts/putExistingContactToDevice/', itemID, '/', userID, '/', dictRoles$'Owner') a <- PUT(link, body = '{"x-auth-token":token}') if ( a$status_code != 200 ){ print(paste0('we are NOT happy! status code -> ', a$status_code)) } else { print(paste('we are happy! status code: ', a$status_code, ' -> ')) } ## ================ B -> remove certain users ================ ## ## analogue to put link <- paste0(sensorURL, 'sensors/contacts/deleteContactFromDevice/', itemID, '/', userID, '/', dictRoles$'Owner') a <- DELETE(link, body = '{"x-auth-token":token}') ## hm, wait for SE... <500> if ( a$status_code != 200 ){ print(paste0('we are NOT happy! status code -> ', a$status_code)) } else { print(paste('we are happy! status code: ', a$status_code, ' -> ')) } ## eof |
input:
https://sensor.awi.de/rest/sensors/contacts/login
--> generate auth tokenhttps://sensor.awi.de/rest/sensors/itemReassignment/assignItemToItem/{itemID}
+ --> content json-bodyCode Block | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|
| ||||||||||
''' basic documentation of sensor.awi.de REST-API can be found here: https://sensor.awi.de/api/ ''' ## necessary libraries import json import requests ## PROD #sensorURL = "https://sensor.awi.de/rest/sensors/" ## SANDBOX sensorURL = "https://sandbox.sensor.awi.de/rest/sensors/" ## + ================================ | xxx | ================================ + ## ''' USE CASE 11: reassign items ''' ## + ================================ | xxx | ================================ + ## ## -------------------------------- preparation -------------------------------- ## ## creating a token userName = 'norbert.anselm@awi.de' PW = '' # <- sensitive information, do not put in git, spaces or elsewhere!!! auth = requests.post(sensorURL + 'contacts/login', data = {'username': userName,'authPassword': PW} ) ## check if the request was successful if auth.status_code == 200: print('we are happy') ## extracting token -> keep it token = auth.cookies['x-auth-token'] else: print('nope, no token available') ## check #answer.status_code ## -------------------------------- reassignment -------------------------------- ## ## I would like to move my newly created thingy ... itemID = 18507 ## ... to my anemometer targetID = 10877 ## point in time when the item was unmounted, deassembled, removed, etc. unmountDate = '2023-01-01T20:00:00' ## point in time when the item was mounted, moved, put to the new environment mountDate = '2023-01-04T12:34:56' body = json.dumps({"targetItemID": targetID, "itemID": itemID, "unmount": unmountDate, "mount": mountDate }) ## doing the job s = requests.put(sensorURL + 'itemReassignment/assignItemToItem/' + str(itemID) , data = body , headers = {"content-type": "application/json"} , cookies = {'x-auth-token': token} ) ## more check s.status_code ## eof |