# coding=utf-8
"""Calculates the montlhy percentiles"""
from calendar import monthrange
from bscearth.utils.log import Log
from earthdiagnostics.diagnostic import Diagnostic, DiagnosticOption, DiagnosticDomainOption, DiagnosticListIntOption
from earthdiagnostics.frequency import Frequencies
from earthdiagnostics.utils import Utils, TempFile
from earthdiagnostics.variable import VariableType
[docs]class MonthlyPercentile(Diagnostic):
"""
Calculates the montlhy percentiles
:param data_manager: data management object
:type data_manager: DataManager
:param startdate: startdate
:type startdate: str
:param member: member number
:type member: int
:param chunk: chunk's number
:type chunk: int
:param variable: variable to average
:type variable: str
"""
alias = 'monpercent'
"Diagnostic alias for the configuration file"
def __init__(self, data_manager, startdate, member, chunk, domain, variable, percentiles):
Diagnostic.__init__(self, data_manager)
self.startdate = startdate
self.member = member
self.chunk = chunk
self.variable = variable
self.domain = domain
self.percentiles = percentiles
def __eq__(self, other):
if self._different_type(other):
return False
return self.startdate == other.startdate and self.member == other.member and self.chunk == other.chunk and \
self.domain == other.domain and self.variable == other.variable and self.percentiles == other.percentiles
def __str__(self):
return 'Monthly percentile Startdate: {0} Member: {1} Chunk: {2} ' \
'Variable: {3}:{4} Percentiles: {5}'.format(self.startdate, self.member, self.chunk,
self.domain, self.variable,
', '.join(str(i) for i in self.percentiles))
[docs] @classmethod
def generate_jobs(cls, diags, options):
"""
Create a job for each chunk to compute the diagnostic
:param diags: Diagnostics manager class
:type diags: Diags
:param options: domain, variable, percentil number, maximum depth (level)
:type options: list[str]
:return:
"""
options_available = (DiagnosticDomainOption(),
DiagnosticOption('variable'),
DiagnosticListIntOption('percentiles', [], 0, 100))
options = cls.process_options(options, options_available)
job_list = list()
for startdate, member, chunk in diags.config.experiment.get_chunk_list():
job_list.append(MonthlyPercentile(diags.data_manager, startdate, member, chunk,
options['domain'], options['variable'], options['percentiles']))
return job_list
[docs] def request_data(self):
"""Request data required by the diagnostic"""
self.variable_file = self.request_chunk(self.domain, self.variable, self.startdate, self.member, self.chunk)
[docs] def declare_data_generated(self):
"""Declare data to be generated by the diagnostic"""
self.max_file = self.declare_chunk(self.domain, self.variable_max, self.startdate, self.member, self.chunk,
frequency=Frequencies.monthly, vartype=VariableType.STATISTIC)
self.min_file = self.declare_chunk(self.domain, self.variable_min, self.startdate, self.member, self.chunk,
frequency=Frequencies.monthly, vartype=VariableType.STATISTIC)
self.percentile_file = {}
for percentile in self.percentiles:
self.percentile_file[percentile] = self.declare_chunk(self.domain, self.percentile(percentile),
self.startdate, self.member, self.chunk,
frequency=Frequencies.monthly,
vartype=VariableType.STATISTIC)
self.declare_chunk(self.domain, '{0}_q{1}'.format(self.variable, percentile), self.startdate,
self.member, self.chunk, frequency=Frequencies.monthly, vartype=VariableType.STATISTIC)
@property
def variable_max(self):
"""
Variable name for the maximum
Returns
-------
str
"""
return '{0}max'.format(self.variable)
@property
def variable_min(self):
"""
Variable name for the minimum
Returns
-------
str
"""
return '{0}min'.format(self.variable)
[docs] def percentile(self, percentile):
"""
Variable name for the given percentile
Parameters
----------
percentile: int
Returns
-------
str
"""
return '{0}_q{1}'.format(self.variable, percentile)
[docs] def compute(self):
"""Run the diagnostic"""
temp = TempFile.get()
handler = Utils.open_cdf(self.variable_file.local_file)
datetimes = Utils.get_datetime_from_netcdf(handler)
handler.close()
start_index = 0
while datetimes[start_index].day != 1 and start_index < handler.size:
start_index += 1
if start_index == datetimes.size:
raise Exception('No complete month for diagnostic {0}'.format(self))
end_index = datetimes.size - 1
while datetimes[end_index].day != monthrange(datetimes[end_index].year, datetimes[end_index].month)[1] \
and end_index >= 0:
end_index -= 1
if end_index < 0:
raise Exception('No complete month for diagnostic {0}'.format(self))
if start_index != 0 or end_index != datetimes.size - 1:
start_date = '{0.year}-{0.month}-{0.day}'.format(datetimes[start_index])
end_date = '{0.year}-{0.month}-{0.day}'.format(datetimes[end_index])
Utils.cdo.seldate('{0},{1}'.format(start_date, end_date), input=self.variable_file.local_file, output=temp)
Utils.rename_variable(temp, 'lev', 'ensemble', False)
else:
Utils.copy_file(self.variable_file.local_file, temp)
Log.debug('Computing minimum')
monmin_file = TempFile.get()
Utils.cdo.monmin(input=temp, output=monmin_file)
Log.debug('Computing maximum')
monmax_file = TempFile.get()
Utils.cdo.monmax(input=temp, output=monmax_file)
for percentile in self.percentiles:
Log.debug('Computing percentile {0}', percentile)
Utils.cdo.monpctl(str(percentile), input=[temp, monmin_file, monmax_file], output=temp)
Utils.rename_variable(temp, 'lev', 'ensemble', False)
handler = Utils.open_cdf(monmax_file)
handler.variables[self.variable].long_name += ' {0} Percentile'.format(percentile)
handler.close()
self.percentiles[percentile].set_local_file(temp, rename_var=self.variable)
Utils.rename_variable(monmax_file, 'lev', 'ensemble', False)
handler = Utils.open_cdf(monmax_file)
handler.variables[self.variable].long_name += ' Monthly Maximum'
handler.close()
self.max_file.set_local_file(monmax_file, rename_var=self.variable)
Utils.rename_variable(monmin_file, 'lev', 'ensemble', False)
handler = Utils.open_cdf(monmin_file)
handler.variables[self.variable].long_name += ' Monthly Minimum'
handler.close()
self.min_file.set_local_file(monmin_file, rename_var=self.variable)