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! |
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]''' 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 |
## ''' ## 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 |
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}
''' 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) |
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-body''' 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 |