# -*- coding: utf-8 -*- """This module does blah blah.""" import requests # from insert_sensor.transactional import insert_sensor from insert_sensor.wrapper import (Offering, FoI, Procedure, SensorType) # import json class Sos(): """ A class to represent a sos service. ... Attributes ---------- sosurl : str first name of the person token : str token to access soso service """ def __init__(self, url, token=''): self.sosurl = str(url) # url to access the SOS self.token = str(token) # security token, optional # Test if URL exists try: test = requests.get(self.sosurl) test.raise_for_status() except requests.HTTPError: print("The URL is not valid") def main(): """ main function """ sos_url = 'https://geomon.geologie.ac.at/52n-sos-webapp/service' # Gschliefgraben Glasfaser # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "inclino1_02", # "Inklinometer inclino1_02, Gschliefgraben Glasfaser" # ) # procedure = Procedure( "inclino1_02","inclino1_02") # foi = FoI("degree", "m", (13.774966, 47.910849, 0.0), # "inclino1-glasfaser-gschliefgraben", # "Glasfaser Untersuchungen am Gschliefgraben (Gmunden)") # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "inclino1_05", # "Inklinometer inclino1_05, Gschliefgraben Glasfaser" # ) # procedure = Procedure("inclino1_05", "inclino1_05") # foi = FoI("degree", "m", (13.774966, 47.910849, 0.0), # "inclino1-glasfaser-gschliefgraben", # "Glasfaser Untersuchungen am Gschliefgraben (Gmunden)") # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "inclino1_14", # "Inklinometer inclino1_14, Gschliefgraben Glasfaser" # ) # procedure = Procedure("inclino1_14", "inclino1_14") # foi = FoI("degree", "m", (13.774966, 47.910849, 0.0), # "inclino1-glasfaser-gschliefgraben", # "Glasfaser Untersuchungen am Gschliefgraben (Gmunden)") offering = Offering( "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", "inclino1_06", "Inklinometer inclino1_06, Gschliefgraben Glasfaser" ) procedure = Procedure("inclino1_06", "inclino1_06") foi = FoI("degree", "m", (13.774966, 47.910849, 0.0), "inclino1-glasfaser-gschliefgraben", "Glasfaser Untersuchungen am Gschliefgraben (Gmunden)") sensor_type = SensorType("inclinometer") post_data = insert_sensor(offering, procedure, foi, sensor_type) print(post_data) headers = {'Accept': 'application/json'} request = requests.post(sos_url, headers=headers, json=post_data) print(request.text) # { # "request" : "InsertSensor", # "version" : "2.0.0", # "service" : "SOS", # "assignedProcedure" : "inclino1_14", # "assignedOffering" : "inclino1_14" # } def insert_sensor(offering, procedure, foi, sensor_type): """ Prepares the body of a InsertSensor request for JSON biding. :param offering: an instance of class Offering.Type object. :param Procedure: instance of class Procedure. type object. :param foi: feature of interest. Instance of FoI :param sensor_type: SensorType object :return: valid body for an InsertSensor request. """ # shortName = offering.name # string # longName = 'Sibratsgfall test' # string # Offering values off_name = '\"' + str(offering.name) + '\"' # Offering name, double quoted offering_name = offering.name offering_label = offering.label # offID = offering.fullId # URL format of full id # featureName = featureID = cordX = cordY = height = h_unit = z_unit = coordinates = "" if foi is not None: # check if feature of interest should be declare # feature_id = 'https://geomon.geologie.ac.at/52n-sos-webapp/api/features/' + \ # str(foi.fid) # URL format cord_x = str(foi.x) # longitude degrees, float cord_y = str(foi.y) # latitude degrees, float coordinates = cord_x + " " + cord_y height = str(foi.z) # altitude in meters, float # h_unit = foi.Hunit # units for horizontal coordinates # z_unit = foi.Vunit # units for altitude feature_id = foi.fid # "feature location" feature_name = foi.name # "feature location" else: pass procedure_name = procedure.name procedure_identifier = procedure.id # URL, obs_types = [] output_list = '' # output list element for describe procedure properties_list = [] for attr in sensor_type.pattern["attributes"]: obs_prop_name = '\"' + attr[0] + '\"' # attribute name # print(obs_prop_name) unit_name = sensor_type.om_types[attr[1]] # om type # magnitud = a # ?? obs_name = obs_prop_name.replace('\"', '') obs_name = "".join(obs_name.split()) # observable property name output = '' output_list = output_list + output # add property identifier to the list. properties_list.append(obs_name) # prepare list of measurement types # A sensor can not registry duplicated sensor types. this_type = "http://www.opengis.net/def/observationType/OGC-OM/2.0/"+unit_name if this_type not in obs_types: # when new type appears obs_types.append(this_type) else: continue # Unit of measurement: unit_name = '\"' + procedure.name + '\"' # double quoted string # unit = omType # one of the MO measurement types body = { "request": "InsertSensor", "service": "SOS", "version": "2.0.0", "procedureDescriptionFormat": "http://www.opengis.net/sensorml/2.0", "procedureDescription": f'{procedure_identifier}shortName{procedure_name}{offering_label}{offering_name}featuresOfInterest{feature_id}{feature_name}{coordinates}Slope{cord_x}{cord_y}{height}', "observableProperty": [ "Slope", # "Roll", # "InSystemTemperature" ], "observationType": [ "http://www.opengis.net/def/observationType/OGC-OM/2.0/OM_Measurement" ], "featureOfInterestType": "http://www.opengis.net/def/samplingFeatureType/OGC-OM/2.0/SF_SamplingPoint" } return body if __name__ == '__main__': main()