Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self):
"""Process initialization"""
# init process
WPSProcess.__init__(self,
identifier = os.path.split(__file__)[-1].split(".")[0],
title="TESTit",
version = 0.1,
abstract = "Just testing basic functions, adding input, checking status, etc...",
storeSupported = "true",
statusSupported = "true",
)
self.valueIn = self.addLiteralInput(identifier = "val", type=types.IntType,title = "User Input Value", default = 6)
self.timePauseIn = self.addLiteralInput(identifier = "pauseTime", type=types.FloatType,title = "User Value for Pause Time", default = 1.)
self.numberPauseIn = self.addLiteralInput(identifier = "pauseNumber", type=types.IntType,title = "User Value for Number of Pauses", default = 3)
self.pauseIn = self.addLiteralInput(identifier = "pause", type=bool,title = "User Input Value", default = True)
self.textOut = self.addLiteralOutput(identifier="text", title="just some text")
def execute(self):
def __init__(self):
# init process
WPSProcess.__init__(self,
identifier = "dummyprocess", # must be same, as filename
title="Dummy Process",
version = "0.1",
storeSupported = "true",
statusSupported = "true",
abstract="The Dummy process is used for testing the WPS structure. The process will accept 2 input numbers and will return the XML result with an add one and subtract one operation",
grassLocation =False)
self.Input1 = self.addLiteralInput(identifier = "input1",
title = "Input1 number",
default="100")
self.Input2= self.addLiteralInput(identifier="input2",
title="Input2 number",
default="200")
self.Output1=self.addLiteralOutput(identifier="output1",
def __init__(self):
"""Process initialization"""
WPSProcess.__init__(self,
identifier=os.path.split(__file__)[-1].split('.')[0],
title='slurm_dispatcher',
version='0.2.1', abstract='Pass esgf compute operation to slurm for processing on cluster',
storeSupported=True,
statusSupported=True)
# this will be operation specific
self.region = self.addComplexInput(identifier='region', title='region of interest and grid specs', formats=[{'mimeType': 'text/json', 'encoding': 'utf-8', 'schema': None}])
# one or more files will be operaton specific
self.dataIn = self.addComplexInput(identifier='data', title='one or more data files', formats=[{'mimeType': 'text/json'}], minOccurs=1, maxOccurs=1)
self.operation = self.addLiteralInput(identifier='operation', type=str, title='operation')
self.result = self.addComplexOutput(identifier='result', title='result of operation', formats=[{'mimeType': 'text/json'}])
def __init__(self):
"""Process initialization"""
WPSProcess.__init__(self, identifier=os.path.split(__file__)[-1].split('.')[0], title='vcs plot', version=0.1, abstract='Generate a plot using vcs', storeSupported='true', statusSupported='true')
self.domain = self.addComplexInput(identifier='domain', title='spatiotemporal domain of plot', formats=[{'mimeType': 'text/json', 'encoding': 'utf-8', 'schema': None}])
# self.download = self.addLiteralInput(identifier='download', type=bool, title='download output', default=False)
self.dataIn = self.addComplexInput(identifier='variable', title='variable to average', formats=[{'mimeType': 'text/json', 'encoding': 'utf-8', 'schema': None}], minOccurs=1, maxOccurs=1)
self.plotSpec = self.addComplexInput(identifier='plot', title='plot specification', formats=[{'mimeType': 'text/json', 'encoding': 'utf-8', 'schema': None}], minOccurs=1, maxOccurs=1)
self.result = self.addLiteralOutput( identifier='result', title='result URL', type=types.StringType )
def __init__(self):
"""Process initialization"""
# init process
WPSProcess.__init__(self,
identifier = os.path.split(__file__)[-1].split(".")[0],
title="ensemble averager",
version = 0.1,
abstract = "Average multiple runs together to create the ensemble mean",
storeSupported = "true",
statusSupported = "true",
)
self.domain = self.addComplexInput(identifier='domain', title='domain over which to average', formats=[{'mimeType': 'text/json', 'encoding': 'utf-8', 'schema': None}])
self.dataIn = self.addComplexInput(identifier = "variable",title="variable to average",formats = [{"mimeType":"text/json"}], minOccurs=1, maxOccurs=10)
self.ensemble = self.addComplexOutput(identifier = "ensemble",title="ensemble average",formats = [{"mimeType":"text/json"}])
def __init__(self):
"""Process initialization"""
WPSProcess.__init__(self, identifier=os.path.split(__file__)[-1].split('.')[0], title='averager', version=0.1, abstract='Average a variable over a (many) dimension', storeSupported='true', statusSupported='true')
self.domain = self.addComplexInput(identifier='domain', title='domain over which to average', formats=[{'mimeType': 'text/json',
'encoding': 'utf-8',
'schema': None}])
self.download = self.addLiteralInput(identifier='download', type=bool, title='download output', default=False)
self.dataIn = self.addComplexInput(identifier='variable', title='variable to average', formats=[{'mimeType': 'text/json'}], minOccurs=1, maxOccurs=1)
self.average = self.addComplexOutput(identifier='average', title='averaged variable', formats=[{'mimeType': 'text/json'}])
def __init__(self):
self.init_time = time.time()
logging.debug("Timeseries Process initialization")
WPSProcess.__init__(self, identifier=os.path.split(__file__)[-1].split('.')[0], title='timeseries', version=0.1, abstract='Extract a timeseries at a spatial location', storeSupported='true', statusSupported='true')
self.domain = self.addComplexInput(identifier='domain', title='spatial location of timeseries', formats=[{'mimeType': 'text/json', 'encoding': 'utf-8', 'schema': None}])
# self.download = self.addLiteralInput(identifier='download', type=bool, title='download output', default=False)
self.dataIn = self.addComplexInput(identifier='variable', title='variable to average', formats=[{'mimeType': 'text/json', 'encoding': 'utf-8', 'schema': None}], minOccurs=1, maxOccurs=1)
self.result = self.addLiteralOutput( identifier='result', title='timeseries data', type=types.StringType )
self.cacheVariableData = False
def __init__(self):
##
# Process initialization
WPSProcess.__init__(self,
identifier = "returner",
title="Return process",
abstract="""This is demonstration process of PyWPS, returns
the same file, it gets on input, as the output.""",
version = "1.0",
storeSupported = "true",
statusSupported = "true")
##
# Adding process inputs
self.dataIn = self.addComplexInput(identifier="data",
title="Input vector data",
formats = [{'mimeType':'text/xml'},{'mimeType': 'image/tiff'}])
self.textIn = self.addLiteralInput(identifier="text",
def __init__(self):
WPSProcess.__init__(self,
identifier = 'wps_percentile_indice', # only mandatory attribute = same file name
title = 'c4i - Percentile Indices',
abstract = 'Computes single input, percentile based indices of temperature: TG10p, TX10p, TN10p, TG90p, TX90p, TN90p, WSDI, CSDI; and of rainfall R75p, R95p, R99p, R75TOT, R95TOT, R99TOT. This processes is also available in Climate4Impact and uses ICCLIM.',
version = "1.0",
metadata = [
{"title": "ICCLIM" , "href": "http://icclim.readthedocs.io/en/latest/"},
{"title": "Climate4Impact", "href": "http://climate4impact.eu/impactportal/general/index.jsp"},
],
storeSupported = True,
statusSupported = True,
grassLocation =False)
## self.filesBasePeriodIn = self.addLiteralInput(identifier = 'filesBasePeriod',
## title = 'Input netCDF files list (base (reference) period)',
## abstract="application/netcdf",
def __init__(self, operation):
""" ESGFProcess init. """
Process.WPSProcess.__init__(
self,
operation.identifier,
operation.title,
abstract=operation.abstract,
version=operation.version,
statusSupported=True,
storeSupported=True)
self.addComplexInput(
'domain',
'Domain',
'Domain the process will utilize.',
metadata=[],
minOccurs=1,
maxOccurs=1,
formats=[