"""Base class for monitoring diagnostics."""
import logging
import os
import re
import cartopy
import matplotlib.pyplot as plt
import yaml
from iris.analysis import MEAN
from mapgenerator.plotting.timeseries import PlotSeries
from esmvaltool.diag_scripts.shared import ProvenanceLogger, names
logger = logging.getLogger(__name__)
def _replace_tags(paths, variable):
"""Replace tags in the config-developer's file with actual values."""
if isinstance(paths, str):
paths = set((paths.strip('/'), ))
else:
paths = set(path.strip('/') for path in paths)
tlist = set()
for path in paths:
tlist = tlist.union(re.findall(r'{([^}]*)}', path))
if 'sub_experiment' in variable:
new_paths = []
for path in paths:
new_paths.extend(
(re.sub(r'(\b{ensemble}\b)', r'{sub_experiment}-\1', path),
re.sub(r'({ensemble})', r'{sub_experiment}-\1', path)))
tlist.add('sub_experiment')
paths = new_paths
for tag in tlist:
original_tag = tag
tag, _, _ = _get_caps_options(tag)
if tag == 'latestversion': # handled separately later
continue
if tag in variable:
replacewith = variable[tag]
else:
raise ValueError(f"Dataset key '{tag}' must be specified for "
f"{variable}, check your recipe entry")
paths = _replace_tag(paths, original_tag, replacewith)
return paths
def _replace_tag(paths, tag, replacewith):
"""Replace tag by replacewith in paths."""
_, lower, upper = _get_caps_options(tag)
result = []
if isinstance(replacewith, (list, tuple)):
for item in replacewith:
result.extend(_replace_tag(paths, tag, item))
else:
text = _apply_caps(str(replacewith), lower, upper)
result.extend(p.replace('{' + tag + '}', text) for p in paths)
return list(set(result))
def _get_caps_options(tag):
lower = False
upper = False
if tag.endswith('.lower'):
lower = True
tag = tag[0:-6]
elif tag.endswith('.upper'):
upper = True
tag = tag[0:-6]
return tag, lower, upper
def _apply_caps(original, lower, upper):
if lower:
return original.lower()
if upper:
return original.upper()
return original
[docs]
class MonitorBase():
"""Base class for monitoring diagnostic.
It contains the common methods for path creation, provenance
recording, option parsing and to create some common plots.
"""
def __init__(self, config):
self.cfg = config
plot_folder = config.get(
'plot_folder',
'{plot_dir}/../../{dataset}/{exp}/{modeling_realm}/{real_name}',
)
plot_folder = plot_folder.replace('{plot_dir}',
self.cfg[names.PLOT_DIR])
self.plot_folder = os.path.abspath(
os.path.expandvars(os.path.expanduser(plot_folder))
)
self.plot_filename = config.get(
'plot_filename',
'{plot_type}_{real_name}_{dataset}_{mip}_{exp}_{ensemble}')
self.plots = config.get('plots', {})
default_config = os.path.join(os.path.dirname(__file__),
"monitor_config.yml")
cartopy_data_dir = config.get('cartopy_data_dir', None)
if cartopy_data_dir:
cartopy.config['data_dir'] = cartopy_data_dir
with open(config.get('config_file', default_config)) as config_file:
self.config = yaml.safe_load(config_file)
def _add_file_extension(self, filename):
"""Add extension to plot filename."""
return f"{filename}.{self.cfg['output_file_type']}"
def _get_proj_options(self, map_name):
return self.config['maps'][map_name]
def _get_variable_options(self, variable_group, map_name):
options = self.config['variables'].get(
variable_group, self.config['variables']['default'])
if 'default' not in options:
variable_options = options
else:
variable_options = options['default']
if map_name in options:
variable_options = {**variable_options, **options[map_name]}
if 'bounds' in variable_options:
if not isinstance(variable_options['bounds'], str):
variable_options['bounds'] = [
float(n) for n in variable_options['bounds']
]
logger.debug(variable_options)
return variable_options
[docs]
def plot_timeseries(self, cube, var_info, period='', **kwargs):
"""Plot timeseries from a cube.
It also automatically smoothes it for long timeseries of monthly data:
- Between 10 and 70 years long, it also plots the 12-month rolling
average along the raw series
- For more than ten years, it plots the 12-month and 10-years
rolling averages and not the raw series
"""
if 'xlimits' not in kwargs:
kwargs['xlimits'] = 'auto'
length = cube.coord("year").points.max() - cube.coord(
"year").points.min()
filename = self.get_plot_path(f'timeseries{period}', var_info,
add_ext=False)
caption = ("{} of "
f"{var_info[names.LONG_NAME]} of dataset "
f"{var_info[names.DATASET]} (project "
f"{var_info[names.PROJECT]}) from "
f"{var_info[names.START_YEAR]} to "
f"{var_info[names.END_YEAR]}.")
if length < 10 or length * 11 > cube.coord("year").shape[0]:
self.plot_cube(cube, filename, **kwargs)
self.record_plot_provenance(
self._add_file_extension(filename),
var_info,
'timeseries',
period=period,
caption=caption.format("Time series"),
)
elif length < 70:
self.plot_cube(cube, filename, **kwargs)
self.record_plot_provenance(
self._add_file_extension(filename),
var_info,
'timeseries',
period=period,
caption=caption.format("Time series"),
)
# Smoothed time series (12-month running mean)
plt.gca().set_prop_cycle(None)
self.plot_cube(cube.rolling_window('time', MEAN, 12),
f"{filename}_smoothed_12_months",
**kwargs)
self.record_plot_provenance(
self._add_file_extension(f"{filename}_smoothed_12_months"),
var_info,
'timeseries',
period=period,
caption=caption.format(
"Smoothed (12-months running mean) time series"),
)
else:
# Smoothed time series (12-month running mean)
self.plot_cube(cube.rolling_window('time', MEAN, 12),
f"{filename}_smoothed_12_months",
**kwargs)
self.record_plot_provenance(
self._add_file_extension(f"{filename}_smoothed_12_months"),
var_info,
'timeseries',
period=period,
caption=caption.format(
"Smoothed (12-months running mean) time series"),
)
# Smoothed time series (10-year running mean)
self.plot_cube(cube.rolling_window('time', MEAN, 120),
f"{filename}_smoothed_10_years",
**kwargs)
self.record_plot_provenance(
self._add_file_extension(f"{filename}_smoothed_10_years"),
var_info,
'timeseries',
period=period,
caption=caption.format(
"Smoothed (10-years running mean) time series"),
)
[docs]
def record_plot_provenance(self, filename, var_info, plot_type, **kwargs):
"""Write provenance info for a given file."""
with ProvenanceLogger(self.cfg) as provenance_logger:
prov = self.get_provenance_record(
ancestor_files=[var_info['filename']],
plot_type=plot_type,
long_names=[var_info[names.LONG_NAME]],
**kwargs,
)
provenance_logger.log(filename, prov)
[docs]
def plot_cube(self, cube, filename, linestyle='-', **kwargs):
"""Plot a timeseries from a cube.
Supports multiplot layouts for cubes with extra dimensions
`shape_id` or `region`.
"""
plotter = PlotSeries()
plotter.filefmt = self.cfg['output_file_type']
plotter.img_template = filename
region_coords = ('shape_id', 'region')
for region_coord in region_coords:
if cube.coords(region_coord):
if cube.coord(region_coord).shape[0] > 1:
plotter.multiplot_cube(cube, 'time', region_coord,
**kwargs)
return
plotter.plot_cube(cube, 'time', linestyle=linestyle, **kwargs)
[docs]
@staticmethod
def get_provenance_record(ancestor_files, **kwargs):
"""Create provenance record for the diagnostic data and plots."""
record = {
'authors': [
'vegas-regidor_javier',
],
'references': [
'acknow_project',
],
'ancestors': ancestor_files,
**kwargs
}
return record
[docs]
def get_plot_path(self, plot_type, var_info, add_ext=True):
"""Get plot full path from variable info.
Parameters
----------
plot_type: str
Name of the plot
var_info: dict
Variable information from ESMValTool
add_ext: bool, optional (default: True)
Add filename extension from configuration file.
"""
return os.path.join(
self.get_plot_folder(var_info),
self.get_plot_name(plot_type, var_info, add_ext=add_ext),
)
[docs]
def get_plot_folder(self, var_info):
"""Get plot storage folder from variable info.
Parameters
----------
var_info: dict
Variable information from ESMValTool
"""
info = {
'real_name': self._real_name(var_info['variable_group']),
**var_info
}
folder = list(_replace_tags(self.plot_folder, info))[0]
if self.plot_folder.startswith('/'):
folder = '/' + folder
if not os.path.isdir(folder):
os.makedirs(folder, exist_ok=True)
return folder
[docs]
def get_plot_name(self, plot_type, var_info, add_ext=True):
"""Get plot filename from variable info.
Parameters
----------
plot_type: str
Name of the plot
var_info: dict
Variable information from ESMValTool
add_ext: bool, optional (default: True)
Add filename extension from configuration file.
"""
info = {
"plot_type": plot_type,
'real_name': self._real_name(var_info['variable_group']),
**var_info
}
file_name = list(_replace_tags(self.plot_filename, info))[0]
if add_ext:
file_name = self._add_file_extension(file_name)
return file_name
@staticmethod
def _set_rasterized(axes=None):
"""Rasterize all artists and collection of axes if desired."""
if axes is None:
axes = plt.gca()
if not isinstance(axes, list):
axes = [axes]
for single_axes in axes:
for artist in single_axes.artists:
artist.set_rasterized(True)
for collection in single_axes.collections:
collection.set_rasterized(True)
@staticmethod
def _real_name(variable_group):
for subfix in ('Ymean', 'Ysum', 'mean', 'sum'):
if variable_group.endswith(subfix):
variable_group = variable_group.replace(subfix, '')
return variable_group
