Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if metric.probabilistic:
raise ValueError(
'probabilistic metric ',
metric.name,
'cannot compute persistence forecast.',
)
# If lead 0, need to make modifications to get proper persistence, since persistence
# at lead 0 is == 1.
if [0] in hind.lead.values:
hind = hind.copy()
with xr.set_options(keep_attrs=True): # keeps lead.attrs['units']
hind['lead'] = hind['lead'] + 1
n, freq = get_lead_cftime_shift_args(hind.lead.attrs['units'], 1)
# Shift backwards shift for lead zero.
hind['init'] = shift_cftime_index(hind, 'init', -1 * n, freq)
# temporarily change `init` to `time` for comparison to verification data time.
hind = hind.rename({'init': 'time'})
inits, verif_dates = return_inits_and_verif_dates(hind, verif, alignment=alignment)
plag = []
for i in hind.lead.values:
a = verif.sel(time=inits[i])
b = verif.sel(time=verif_dates[i])
a['time'] = b['time']
plag.append(
metric.function(a, b, dim='time', comparison=__e2c, **metric_kwargs)
)
pers = xr.concat(plag, 'lead')
pers['lead'] = hind.lead.values
# keep coords from hind
def _same_inits_alignment(init_lead_matrix, valid_inits, all_verifs, leads, n, freq):
"""Returns initializations and verification dates, maintaining a common set of inits
at all leads.
See ``return_inits_and_verif_dates`` for descriptions of expected variables.
"""
verifies_at_all_leads = init_lead_matrix.isin(all_verifs).all('lead')
inits = valid_inits.where(verifies_at_all_leads, drop=True)
inits = {lead: inits for lead in leads}
verif_dates = {
lead: shift_cftime_index(inits[lead], 'time', n, freq)
for (lead, n) in zip(leads, n)
}
return inits, verif_dates
lw=lw,
hue='member',
color='gray',
alpha=member_alpha,
label='uninitialized',
zorder=hind.lead.size + 1,
)
for i, lead in enumerate(hind.lead.values):
h = hind.sel(lead=lead).rename({'init': 'time'})
if not show_members and 'member' in h.dims:
h = h.mean('member')
lead_alpha = 1
else:
lead_alpha = 0.5
h['time'] = shift_cftime_index(h.time, 'time', int(lead), lead_freq)
h.plot(
ax=ax,
hue='member',
color=cmap(i),
label=f'initialized: lead={lead}',
alpha=lead_alpha,
zorder=hind.lead.size - i,
)
linestyles = ['-', ':', '-.', '--']
if len(obs) > len(linestyles):
raise ValueError(f'Please provide fewer than {len(linestyles)+1} observations.')
if len(obs) > 0:
for i, (obs_name, obs_item) in enumerate(obs.items()):
if isinstance(obs_item, xr.Dataset):
obs_item = obs_item[variable]
n (tuple of ints): Number of units to shift for ``leads``. ``value`` for
``CFTimeIndex.shift(value, str)``.
freq (str): Pandas frequency alias. ``str`` for
``CFTimeIndex.shift(value, str)``.
leads (list, array, xr.DataArray of ints): Leads to return offset for.
Returns:
init_lead_matrix (``xr.DataArray``): DataArray with x=inits and y=lead with
values corresponding to "real time", or ``init + lead`` over all inits and
leads.
"""
# Note that `init` is renamed to `time` in compute functions.
init_lead_matrix = xr.concat(
[
xr.DataArray(
shift_cftime_index(forecast, 'time', n, freq),
dims=['time'],
coords=[forecast['time']],
)
for n in n
],
'lead',
)
init_lead_matrix['lead'] = leads
return init_lead_matrix
start_str = i.strftime()[:10]
if initialized.lead.min() == 0:
dsi['time'] = xr.cftime_range(
start=start_str,
freq=lead_freq,
periods=dsi.time.size,
calendar=calendar,
)
elif initialized.lead.min() == 1:
dsi['time'] = xr.cftime_range(
start=start_str,
freq=lead_freq,
periods=dsi.time.size,
calendar=calendar,
)
dsi['time'] = shift_cftime_index(dsi.time, 'time', 1, lead_freq)
if uninitialized_present:
dsu['time'] = dsi['time']
if not show_members:
dsi = dsi.mean('member')
if uninitialized_present:
dsu = dsu.mean('member')
member_alpha = 1
lw = 2
labelstr = 'ensemble mean'
else:
member_alpha = 0.5
lw = 1
labelstr = 'members'
# plot ensemble mean, first white then color to highlight ensemble mean
if uninitialized_present:
dsu.mean('member').plot(ax=ax, color='white', lw=3, zorder=8, alpha=0.6)