# -*- coding: utf-8 -*- """This module does blah blah.""" from ast import List import requests # from insert_sensor.transactional import insert_sensor from insert_sensor.wrapper import (Offering, FoI, Procedure, SensorType) # import json class Sensor: """ A class to represent an input sensor. ... Attributes ---------- name : str first name of the person x : float token to access soso service y : float token to access soso service """ def __init__(self, name: str, x_coord: float, y_coord: float, feature_id: str, feature_name: str): self.name = name self.x_coord = x_coord self.y_coord = y_coord self.feature_id = feature_id self.feature_name = feature_name def main(): """ main function """ sos_url = 'https://geomon.geologie.ac.at/52n-sos-webapp/service' # Gschliefgraben Piezometer # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "bohrloch1", # "Bohrlöcher, Gschliefgraben Piezometer" # ) # procedure = Procedure("bohrloch1", "bohrloch1") # foi = FoI("degree", "m", (13.774966, 47.910849, 0.0), # "bohrloch1-glasfaser-gschliefgraben", # "Piezometer1 am Gschliefgraben") # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "bohrloch2", # "Bohrloch2, Gschliefgraben Piezometer" # ) # procedure = Procedure("bohrloch2", "bohrloch2") # foi = FoI("degree", "m", (13.80957276439, 47.882524348741, 0.0), # "bohrloch2-glasfaser-gschliefgraben", # "Piezometer2 am Gschliefgraben") # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "bohrloch3", # "Bohrloch, Gschliefgraben Piezometer" # ) # procedure = Procedure("bohrloch3", "bohrloch3") # foi = FoI("degree", "m", (13.809990909737, 47.882824994038, 0.0), # "bohrloch3-glasfaser-gschliefgraben", # "Piezometer3 am Gschliefgraben") # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "bohrloch4", # "Bohrloch, Gschliefgraben Piezometer" # ) # procedure = Procedure("bohrloch4", "bohrloch4") # foi = FoI("degree", "m", (13.809379587392, 47.883098856837, 0.0), # "bohrloch4-glasfaser-gschliefgraben", # "Piezometer4 am Gschliefgraben") # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "bohrloch5", # "Bohrloch, Gschliefgraben Piezometer" # ) # procedure = Procedure("bohrloch5", "bohrloch5") # foi = FoI("degree", "m", (13.81120655331, 47.884145740545, 0.0), # "bohrloch5-glasfaser-gschliefgraben", # "Piezometer5 am Gschliefgraben") # offering = Offering( # "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", # "bohrloch6", # "Bohrloch, Gschliefgraben Piezometer" # ) # procedure = Procedure("bohrloch6", "bohrloch6") # foi = FoI("degree", "m", (13.811537883268, 47.885082327907, 0.0), # "bohrloch6-glasfaser-gschliefgraben", # "Piezometer6 am Gschliefgraben") # creating list sensor_list: List[Sensor] = [] # appending instances to list sensor_list.append( Sensor('gschliefgraben_bohrloch_1', 13.774966, 47.910849, "bohrloch1-glasfaser-gschliefgraben", "Piezometer1 am Gschliefgraben")) sensor_list.append( Sensor('gschliefgraben_bohrloch_2', 13.80957276439, 47.882524348741, "bohrloch2-glasfaser-gschliefgraben", "Piezometer2 am Gschliefgraben")) sensor_list.append( Sensor('gschliefgraben_bohrloch_3', 13.809990909737, 47.882824994038, "bohrloch3-glasfaser-gschliefgraben", "Piezometer3 am Gschliefgraben")) sensor_list.append( Sensor('gschliefgraben_bohrloch_4', 13.809379587392, 47.883098856837, "bohrloch4-glasfaser-gschliefgraben", "Piezometer4 am Gschliefgraben")) sensor_list.append( Sensor('gschliefgraben_bohrloch_5', 13.81120655331, 47.884145740545, "bohrloch5-glasfaser-gschliefgraben", "Piezometer5 am Gschliefgraben")) sensor_list.append( Sensor('gschliefgraben_bohrloch_6', 13.811537883268, 47.885082327907, "bohrloch6-glasfaser-gschliefgraben", "Piezometer6 am Gschliefgraben")) sensor: Sensor for sensor in sensor_list: # platform wolfsegg offering = Offering( "https://geomon.geologie.ac.at/52n-sos-webapp/api/offerings/", sensor.name, "Bohrloch, Gschliefgraben Piezometer" ) procedure = Procedure(sensor.name, sensor.name) foi = FoI("degree", "m", (sensor.x_coord, sensor.y_coord, 0.0), sensor.feature_id, sensor.feature_name) # now insert sensor via rest service: sensor_type = SensorType("piezometer") 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" : "bohrloch1", # "assignedOffering" : "bohrloch" # } 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 gml_id = '\"' + str(procedure.id) + '\"' # 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 = "" # check if feature of interest should be declare if foi is not None: # 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}longName{procedure_name}shortName{procedure_name}{offering_label}{offering_name}truefalsefeaturesOfInterest{feature_id}{feature_name}{coordinates}Elevation{cord_x}{cord_y}{height}', "observableProperty": [ "Elevation", # "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()