Source code for solarforecastarbiter.plotting.timeseries
import datetime as dt
from functools import wraps
import logging
from bokeh.embed import components
from bokeh.layouts import gridplot
from bokeh.models import ColumnDataSource, Label, HoverTool
from bokeh.plotting import figure
from bokeh import palettes
from matplotlib import cm
from matplotlib.colors import Normalize
import plotly.graph_objects as go
import pandas as pd
import pytz
from solarforecastarbiter.plotting import utils as plot_utils
from solarforecastarbiter.validation import quality_mapping
UTCS = (dt.timezone.utc, pytz.UTC)
logger = logging.getLogger('sfa.plotting.timeseries')
PLOT_WIDTH = 900
PALETTE = palettes.all_palettes['Category20b'][20][::4]
# flags with color None will be assigned a color from PALETTE
FLAG_COLORS = {
'MISSING': '#e6550d',
'NOT VALIDATED': '#ff7f0e',
'USER FLAGGED': '#d62728',
'NIGHTTIME': None,
'DAYTIME': None,
'CLEARSKY': None,
'SHADED': None,
'UNEVEN FREQUENCY': None,
'LIMITS EXCEEDED': None,
'CLEARSKY EXCEEDED': None,
'STALE VALUES': None,
'DAYTIME STALE VALUES': None,
'INTERPOLATED VALUES': None,
'DAYTIME INTERPOLATED VALUES': None,
'CLIPPED VALUES': None,
'INCONSISTENT IRRADIANCE COMPONENTS': None,
}
[docs]def build_figure_title(object_name, start, end):
"""Builds a title for the plot
Parameters
----------
object_name : str
Name of the object being plotted
start: datetime-like
The start of the interval being plotted.
end: datetime-like
The end of the interval being plotted.
Returns
-------
string
The appropriate figure title.
Raises
------
ValueError
If start or end is not localized to UTC
"""
if start.tzinfo not in UTCS or end.tzinfo not in UTCS:
raise ValueError('start and end must be localized to UTC')
start_string = start.strftime('%Y-%m-%d %H:%M')
end_string = end.strftime('%Y-%m-%d %H:%M')
figure_title = (f'{object_name} {start_string} to {end_string} UTC')
return figure_title
def _single_quality_bar(flag_name, plot_width, x_range, color, source):
qfig = figure(sizing_mode='stretch_width',
plot_height=30,
plot_width=plot_width,
x_range=x_range,
toolbar_location=None,
min_border_bottom=0,
min_border_top=0,
tools='xpan',
x_axis_location=None,
y_axis_location=None)
qfig.ygrid.grid_line_color = None
qfig.line(x='timestamp', y=flag_name,
line_width=qfig.plot_height,
source=source, alpha=0.6,
line_color=color)
flag_label = Label(x=5, y=0,
x_units='screen', y_units='screen',
text=flag_name, render_mode='css',
border_line_color=None,
background_fill_alpha=0,
text_font_size='1em',
text_font_style='bold')
qfig.add_layout(flag_label)
return qfig
[docs]def make_quality_bars(source, plot_width, x_range):
"""
Make figures to display the whether a time is flagged for any
of the columns in source.
Parameters
----------
source : bokeh.models.ColumnDataSource
The predefined data source with flags loaded. Only columns
in FLAG_COLORS will be made into bars. If data for a flag
is empty, a bar will not be generated for that flag.
plot_width : int
The width of the figures
x_range : bokeh.Range or tuple
If x_range is a bokeh Range from another plot, the plots will
be linked on panning/zooming/etc.
Returns
-------
list
Of bar figures. The top figure will have an appropriate title.
"""
palette = iter(PALETTE * 3)
out = []
for flag, color in FLAG_COLORS.items():
# only display bars for flags that have at least on occurence
if flag not in source.data or pd.isna(source.data[flag]).all():
continue
if color is None:
color = next(palette)
nextfig = _single_quality_bar(flag, plot_width, x_range,
color, source)
out.append(nextfig)
# add the title to the top bar if there are any bars
if out:
out[0].plot_height = 60
out[0].title.text = 'Quality Flags'
out[0].title.text_font_size = '1em'
return out
[docs]def add_hover_tool(fig, source, **hover_kwargs):
"""Add a hover tool to fig. If `add_line=True` in `hover_kwargs`
an invisible line is added to enable hover values step plots"""
if hover_kwargs.pop('add_line', False):
fig.line(x='timestamp', y='value', source=source, line_alpha=0)
tooltips = [
('timestamp', '@timestamp{%FT%H:%M:%S%z}'),
('value', '@value{%0.2f}')
]
if 'active_flags' in source.data.keys():
tooltips.append(('quality flags', '@active_flags'))
hover = HoverTool(tooltips=tooltips, formatters={
'timestamp': 'datetime', 'value': 'printf'}, mode='vline',
**hover_kwargs)
fig.add_tools(hover)
[docs]def make_basic_timeseries(source, object_name, variable, interval_label,
plot_width):
"""
Make a basic timeseries plot (with either a step or line)
and add a hover tool.
Parameters
----------
source : bokeh.models.ColumnDataSource
The datasource with 'timestamp' and 'value' columns that will be
plotted.
object_name : str
Name of the object to be plotted so an appropriate title can be made.
variable : str
Variable of the plotted object
interval_label : str
Interval label of the object to determine whether a line or step
is most appropriate.
plot_width : int
The width of the output figure
Returns
-------
bokeh.models.Figure
The figure with the timeseries
Raises
------
KeyError
If timestamp is not a column in source
IndexError
If the timestamp column is empty
"""
timestamps = source.data['timestamp']
first = pd.Timestamp(timestamps[0], tz='UTC')
last = pd.Timestamp(timestamps[-1], tz='UTC')
plot_method, plot_kwargs, hover_kwargs = plot_utils.line_or_step(
interval_label)
figure_title = build_figure_title(object_name, first, last)
fig = figure(title=figure_title, sizing_mode='scale_width',
plot_width=plot_width,
plot_height=300, x_range=(first, last),
x_axis_type='datetime',
tools='pan,wheel_zoom,box_zoom,zoom_in,zoom_out,reset,save',
toolbar_location='above',
min_border_bottom=50)
getattr(fig, plot_method)(x='timestamp', y='value', source=source,
**plot_kwargs)
fig.yaxis.axis_label = plot_utils.format_variable_name(variable)
fig.xaxis.axis_label = 'Time (UTC)'
if variable == 'event':
fig.yaxis.ticker = [0, 1]
fig.yaxis.major_label_overrides = {1: 'True', 0: 'False'}
add_hover_tool(fig, source, **hover_kwargs)
return fig
def _make_layout(figs):
layout = gridplot(figs,
ncols=1,
merge_tools=True,
toolbar_location='above',
sizing_mode='scale_width')
return layout
def to_components(f):
"""Return script and div of a bokeh object if the return_components
kwarg is True"""
@wraps(f)
def wrapper(*args, **kwargs):
if kwargs.pop('return_components', False):
out = f(*args, **kwargs)
if out is not None:
return components(out)
else:
return out
else:
return f(*args, **kwargs)
return wrapper
[docs]@to_components
def generate_forecast_figure(forecast, data, limit=None):
"""
Creates a bokeh timeseries figure for forcast data
Parameters
----------
forecast : datamodel.Forecast
The Forecast that is being plotted
data : pandas.Series
The forecast data with a datetime index to be plotted
limit : pandas.Timedelta or None
The time limit from the last datapoint to plot. If None, all
data is plotted.
Returns
-------
None
When the data is empty
script, div : str
When return_components = True, return the <script> and <div>
components for the Bokeh plot.
bokeh components from gridplot
When return_components = False
"""
logger.info('Starting forecast figure generation...')
if len(data.index) == 0:
return None
data = plot_utils.align_index(data, forecast.interval_length, limit)
cds = ColumnDataSource(data.reset_index())
fig = make_basic_timeseries(cds, forecast.name, forecast.variable,
forecast.interval_label, PLOT_WIDTH)
layout = _make_layout([fig])
logger.info('Figure generated succesfully')
return layout
[docs]@to_components
def generate_observation_figure(observation, data, limit=pd.Timedelta('3d')):
"""
Creates a bokeh figure from API responses for an observation
Parameters
----------
observation : datamodel.Observation
The Observation that is being plotted
data : pandas.DataFrame
The observation data to be plotted with datetime index
and ('value', 'quality_flag') columns
limit : pandas.Timedelta or None
The time limit from the last datapoint to plot. If None, all
data is plotted.
Returns
-------
None
When the data is empty
script, div : str
When return_components = True, return the <script> and <div>
components for the Bokeh plot.
bokeh components from gridplot
When return_components = False
"""
logger.info('Starting observation forecast generation...')
if len(data.index) == 0:
return None
data = plot_utils.align_index(data, observation.interval_length, limit)
quality_flag = data.pop('quality_flag').dropna().astype(int)
bool_flags = quality_mapping.convert_mask_into_dataframe(quality_flag)
active_flags = quality_mapping.convert_flag_frame_to_strings(bool_flags)
active_flags.name = 'active_flags'
flags = bool_flags.mask(~bool_flags).reindex(data.index) # add missing
flags['MISSING'] = pd.Series(1.0, index=data.index)[pd.isna(data['value'])]
# need to fill as line needs more than a single point to show up
if observation.interval_label == 'ending':
flags.bfill(axis=0, limit=1, inplace=True)
else:
# for interval beginning and instantaneous
flags.ffill(axis=0, limit=1, inplace=True)
cds = ColumnDataSource(pd.concat([data, flags, active_flags], axis=1))
figs = [make_basic_timeseries(cds, observation.name, observation.variable,
observation.interval_label, PLOT_WIDTH)]
figs.extend(make_quality_bars(cds, PLOT_WIDTH, figs[0].x_range))
layout = _make_layout(figs)
logger.info('Figure generated succesfully')
return layout
PLOTLY_MARGINS = {'l': 50, 'r': 50, 'b': 50, 't': 100, 'pad': 4}
PLOTLY_LAYOUT_DEFAULTS = {
'autosize': True,
'height': 300,
'margin': PLOTLY_MARGINS,
'plot_bgcolor': '#FFF',
'title_font_size': 16,
'font': {'size': 14}
}
def _plot_probabilsitic_distribution_axis_y(fig, forecast, data):
"""
Plot all probabilistic forecast values for axis='y' by adding traces to
fig.
Parameters
----------
fig: plotly.graph_objects.Figure
forecast: :py:class`solarforecastarbiter.datamodel.ProbabilisticForecast`
data: pd.DataFrame
"""
color_map = cm.get_cmap('viridis')
color_scaler = cm.ScalarMappable(
Normalize(vmin=0, vmax=1),
color_map,
)
units = forecast.units
percentiles_are_symmetric = plot_utils.percentiles_are_symmetric(
data.columns.values.astype('float'))
# may not work for constant values that don't convert nicely from str/float
constant_values = data.columns.astype('float').sort_values()
for i, constant_value in enumerate(constant_values):
if i == 0:
fill = None
else:
fill = 'tonexty'
if percentiles_are_symmetric:
if constant_value <= 50 and i != 0:
fill_value = constant_values[i - 1]
else:
fill_value = constant_value
fill_value = 2 * abs(fill_value - 50)
else:
fill_value = 100 - constant_value
fill_color = plot_utils.distribution_fill_color(
color_scaler, fill_value)
plot_kwargs = plot_utils.line_or_step_plotly(forecast.interval_label)
forecast_name = f'Prob(f <= x) = {str(constant_value)}%'
go_ = go.Scatter(
x=data.index,
y=data[str(constant_value)],
name=f'{str(constant_value)} %',
hovertemplate=(
f'<b>{forecast_name}</b><br>'
'<b>Value</b>: %{y} '+f'{units}<br>'
'<b>Time</b>: %{x}<br>'),
connectgaps=False,
showlegend=False,
mode='lines',
fill=fill,
fillcolor=fill_color,
line=dict(
color=fill_color,
),
**plot_kwargs,
)
fig.add_trace(go_)
def _plot_probabilsitic_distribution_axis_x(fig, forecast, data):
"""
Plot all probabilistic forecast values for axis='x' by adding traces to
fig.
Parameters
----------
fig: plotly.graph_objects.Figure
forecast: :py:class`solarforecastarbiter.datamodel.ProbabilisticForecast`
data: pd.DataFrame
"""
palette = iter(PALETTE * 3)
units = forecast.units
for constant_value in data.columns:
line_color = next(palette)
plot_kwargs = plot_utils.line_or_step_plotly(forecast.interval_label)
forecast_name = f'Prob(x <= {str(constant_value)} {units})'
go_ = go.Scatter(
x=data.index,
y=data[str(constant_value)],
name=forecast_name,
hovertemplate=(
f'<b>{forecast_name}</b><br>'
'<b>Value</b>: %{y} %<br>'
'<b>Time</b>: %{x}<br>'),
connectgaps=False,
showlegend=True,
mode='lines',
line=dict(
color=line_color,
),
**plot_kwargs,
)
fig.add_trace(go_)
[docs]def generate_probabilistic_forecast_figure(
forecast, data, limit=pd.Timedelta('3d')):
"""
Creates a plotly figure spec from api response for a probabilistic forecast
group.
Parameters
----------
forecast : datamodel.ProbabilisticForecast
data : pandas.DataFrame
DataFrame with forecast values in each column, column names as the
constant values and a datetime index.
limit : pandas.Timedelta or None
Returns
-------
None
When the data is empty.
figure: Plotly.graph_objects.Figure
Plotly json specification for the plot.
"""
logger.info('Starting probabilistic forecast figure generation...')
if len(data.index) == 0:
return None
fig = go.Figure()
if 'x' in forecast.axis:
ylabel = 'Probability (%)'
_plot_probabilsitic_distribution_axis_x(fig, forecast, data)
else:
ylabel = plot_utils.format_variable_name(forecast.variable)
_plot_probabilsitic_distribution_axis_y(fig, forecast, data)
fig.update_xaxes(title_text=f'Time (UTC)', showgrid=True,
gridwidth=1, gridcolor='#CCC', showline=True,
linewidth=1, linecolor='black', ticks='outside')
fig.update_yaxes(title_text=ylabel, showgrid=True,
gridwidth=1, gridcolor='#CCC', showline=True,
linewidth=1, linecolor='black', ticks='outside',
fixedrange=True)
first = data.index[0]
last = data.index[-1]
fig.update_layout(
title=build_figure_title(forecast.name, first, last),
legend=dict(font=dict(size=10)),
**PLOTLY_LAYOUT_DEFAULTS,
)
return fig