gCubeSystem
/
geoportal-esquilino-sample-python-client
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
geoportal-esquilino-sample-.../main.py

240 lines
9.1 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# @author: Francesco Mangiacrapa
#
# Created on 2023/02/16
#
import requests
from JSON_Reader import JSON_Reader
from IAM_Token_Client import IAM_Token_Client
# Test files URLs are located here
folder_test_files_url = 'https://code-repo.d4science.org/gCubeSystem/geoportal-esquilino-sample-python-client/raw/branch/master/test_files/'
dataset_json_filename = 'test_dataset_esquilino.json'
pdf_filename = 'file-example_PDF_1MB.pdf'
image_filenames = ['arco-di-san-vito.jpg', 'piazza_esquilino.jpg']
shapefile_filename = 'area_di_scavo.shp'
# The UCD ID
UCD_ID = 'esquilino'
# The KC client UMA TOKEN
service_Client_ID = ''
service_Client_Secret = ''
gcube_context = '/d4science.research-infrastructures.eu/gCubeApps/Esquiline'
# EndPoints
# D4Science IAM
IAM_URL = 'https://accounts.d4science.org/auth/realms/d4science/protocol/openid-connect/token'
# The Geoportal Service endpoint
geoportal_service_endpoint = 'https://geoportal-cms.int.d4science.net//geoportal-service/srv'
# Request Authorization (UMA) TOKEN to D4Science IAM for service_Client_ID/service_Client_Secret
def request_UMA_Token():
itr = IAM_Token_Client(IAM_URL, service_Client_ID, service_Client_Secret)
service_uma_token = itr.getUmaToken(gcube_context)
# print("Uma token is: %s" % service_uma_token)
return service_uma_token
# Init UMA Token
uma_token_field = "Bearer " + request_UMA_Token()
class Geoportal_Client_Esquilino():
def __init__(self):
self.init_sample_filesets_for_esquilino()
def init_sample_filesets_for_esquilino(self):
self.fileset_fonteInformazione = {
"toSetAccess": {
"_policy": "OPEN",
"_license": "CCBYSA.4.0"
},
"clashOption": "REPLACE_EXISTING",
"fieldDefinitionPath": "$.fonteInformazione._children[?(@.filesetIta)]",
"parentPath": "$.fonteInformazione",
"fieldName": "filesetIta",
"streams": [{"url": folder_test_files_url + pdf_filename,
"filename": pdf_filename}]
}
self.fileset_documentazioneScientifica = {
"toSetAccess": {
"_policy": "OPEN",
"_license": "CCBYSA.4.0"
},
"clashOption": "REPLACE_EXISTING",
"fieldDefinitionPath": "$.documentazioneScientifica._children[?(@.fileset)]",
"parentPath": "$.documentazioneScientifica",
"fieldName": "fileset",
"streams": [{"url": folder_test_files_url + pdf_filename,
"filename": pdf_filename}]
}
self.fileset_datiAltimetrici = {
"toSetAccess": {
"_policy": "OPEN",
"_license": "CCBYSA.4.0"
},
"clashOption": "REPLACE_EXISTING",
"fieldDefinitionPath": "$.datiAltimetrici._children[?(@.fileset)]",
"parentPath": "$.datiAltimetrici",
"fieldName": "fileset",
"streams": [{"url": folder_test_files_url + shapefile_filename,
"filename": shapefile_filename}]
}
self.fileset_planimetria = {
"toSetAccess": {
"_policy": "OPEN",
"_license": "CCBYSA.4.0"
},
"clashOption": "REPLACE_EXISTING",
"fieldDefinitionPath": "$.planimetria._children[?(@.fileset)]",
"parentPath": "$.planimetria",
"fieldName": "fileset",
"streams": [{"url": folder_test_files_url + shapefile_filename,
"filename": shapefile_filename}]
}
def build_step(self, step_id):
step = {
"stepID": step_id
}
return step
# immaginiRappresentative' is managed as an Array by definition in the Esquilino UCD Model
def build_image_fileset(self, image_index, image_filename):
fileset_immagineRappresentativa = {
"toSetAccess": {
"_policy": "OPEN",
"_license": "CCBYSA.4.0"
},
"clashOption": "REPLACE_EXISTING",
"fieldDefinitionPath": "$.immaginiRappresentative._children[?(@.fileset)]",
"parentPath": "$.immaginiRappresentative[{}]".format(image_index),
"fieldName": "fileset",
"streams": [{"url": folder_test_files_url + image_filename,
"filename": image_filename}]
}
return fileset_immagineRappresentativa
def get_ucd(self, ucd_id):
print("\nget_ucd()")
request_url = geoportal_service_endpoint + "/ucd/" + ucd_id
payload = ""
headers = {
'Content-Type': 'application/json; charset=utf-8',
'Authorization': uma_token_field
}
print("calling: " + request_url)
return requests.request("GET", request_url, headers=headers, data=payload)
def get_project(self, ucd_id, project_id):
print("\nget_project()")
request_url = geoportal_service_endpoint + "/projects/" + ucd_id + "/" + project_id
payload = ""
headers = {
'Content-Type': 'application/json; charset=utf-8',
'Authorization': uma_token_field
}
print("calling: " + request_url)
return requests.request("GET", request_url, headers=headers, data=payload)
def create_project(self, ucd_id, json_data):
print("\ncreate_project()")
request_url = geoportal_service_endpoint + "/projects/" + ucd_id
headers = {
'Content-Type': 'application/json; charset=utf-8',
'Authorization': uma_token_field
}
print("calling: " + request_url)
print("json_data: " + str(json_data))
return requests.post(request_url, headers=headers, json=json_data)
def registrer_fileset(self, ucd_id, project_id, json_data):
print("\nregistrer_fileset()")
request_url = geoportal_service_endpoint + "/projects/" + ucd_id + "/registerFiles/" + project_id
headers = {
'Content-Type': 'application/json; charset=utf-8',
'Authorization': uma_token_field
}
print("calling: " + request_url)
print("json_data: " + str(json_data))
return requests.post(request_url, headers=headers, json=json_data)
def perform_step(self, ucd_id, project_id, json_data):
print("\nperform_step()")
request_url = geoportal_service_endpoint + "/projects/" + ucd_id + "/step/" + project_id
headers = {
'Content-Type': 'application/json; charset=utf-8',
'Authorization': uma_token_field
}
print("calling: " + request_url)
print("json_data: " + str(json_data))
return requests.post(request_url, headers=headers, json=json_data)
def main():
json_esquilino_url = folder_test_files_url + dataset_json_filename;
json_reader = JSON_Reader(json_esquilino_url)
print("\n\nREAD TEST ESQUILINO JSON")
JSON_Reader.pretty_print_json(json_reader.data, 3)
gce = Geoportal_Client_Esquilino()
# print("\n\nRead UCD for id "+UCD_ID)
# response = gce.get_ucd(UCD_ID)
# print("\n\nRequest UCD response for %s" % UCD_ID)
# print("Response code is %s" % response.status_code)
# JSON_Reader.pretty_print_json(response.json(), 3)
print("\n\nCreate a Project for %s" % UCD_ID)
response = gce.create_project(UCD_ID, json_reader.data)
print("Response code is %s" % response.status_code)
print("Created the Project, response is: %s" % response.json())
# Reading the project_id from the response (just created Project)
json_data = JSON_Reader.to_json(response.text)
project_id = json_data["_id"]
# Registering FILESETS for
# Fonte Informazione
response = gce.registrer_fileset(UCD_ID, project_id, gce.fileset_fonteInformazione)
print("Response code is %s" % response.status_code)
# Documentazione Scientifica
response = gce.registrer_fileset(UCD_ID, project_id, gce.fileset_documentazioneScientifica)
print("Response code is %s" % response.status_code)
# Immagini
image_1 = gce.build_image_fileset(0, image_filenames[0])
response = gce.registrer_fileset(UCD_ID, project_id, image_1)
print("Response code is %s" % response.status_code)
image_2 = gce.build_image_fileset(1, image_filenames[1])
response = gce.registrer_fileset(UCD_ID, project_id, image_2)
print("Response code is %s" % response.status_code)
# Dati Altimetrici
response = gce.registrer_fileset(UCD_ID, project_id, gce.fileset_datiAltimetrici)
print("Response code is %s" % response.status_code)
# Planimetria
response = gce.registrer_fileset(UCD_ID, project_id, gce.fileset_planimetria)
print("Response code is %s" % response.status_code)
# Performing STEP PUBLISH
step_publish = gce.build_step("PUBLISH")
response = gce.perform_step(UCD_ID, project_id, step_publish)
print("Response code is %s" % response.status_code)
print("\n\nRead the Project for ucd " + UCD_ID + " and project " + project_id)
response = gce.get_project(UCD_ID, project_id)
print("Response code is %s" % response.status_code)
print("\n\nRead Project response %s" % response.json())
main()
Reference in New Issue View Git Blame Copy Permalink