Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
series = series.ffill(limit=limit)
elif fill_method == 'bfill':
series = series.bfill(limit=limit)
elif fill_method == 'nearest':
if limit == 0:
# limit=0 is a trick to only use each measurements once
# only keep days with measurements
series = series.dropna()
# generate an index at the 14th and 28th of every month
buf = pd.to_timedelta(8, 'd')
ref_index = pd.date_range(series.index.min() - buf,
series.index.max() + buf)
mask = [(x.day == 14) or (x.day == 28) for x in ref_index]
ref_index = ref_index[mask]
# only keep the days that are closest to series.index
ref_index = get_sample(ref_index, series.index)
# and set the index of series to this index
# (and remove rows in series that are not in ref_index)
series = series.reindex(ref_index, method=fill_method)
select14or28 = False
else:
# with a large limit (larger than 6) it is possible that one measurement is used more than once
series = series.dropna().reindex(series.index, method=fill_method,
limit=limit)
else:
series = series.interpolate(method=fill_method, limit=limit,
limit_direction='both')
# and select the 14th and 28th of each month (if needed still)
if select14or28:
mask = [(x.day == 14) or (x.day == 28) for x in series.index]
series = series.loc[mask]
series = series.ffill(limit=limit)
elif fill_method == 'bfill':
series = series.bfill(limit=limit)
elif fill_method == 'nearest':
if limit == 0:
# limit=0 is a trick to only use each measurements once
# only keep days with measurements
series = series.dropna()
# generate an index at the 14th and 28th of every month
buf = to_timedelta(8, 'd')
ref_index = date_range(series.index.min() - buf,
series.index.max() + buf)
mask = [(x.day == 14) or (x.day == 28) for x in ref_index]
ref_index = ref_index[mask]
# only keep the days that are closest to series.index
ref_index = get_sample(ref_index, series.index)
# and set the index of series to this index
# (and remove rows in series that are not in ref_index)
series = series.reindex(ref_index, method=fill_method)
select14or28 = False
else:
# with a large limit (larger than 6) it is possible that one measurement is used more than once
series = series.dropna().reindex(series.index, method=fill_method,
limit=limit)
else:
series = series.interpolate(method=fill_method, limit=limit,
limit_direction='both')
# and select the 14th and 28th of each month (if needed still)
if select14or28:
mask = [(x.day == 14) or (x.day == 28) for x in series.index]
series = series.loc[mask]
update_observations = False
for key, setting in zip([tmin, tmax, freq], ["tmin", "tmax", "freq"]):
if key != self.settings[setting]:
update_observations = True
if self.oseries_calib is None or update_observations:
oseries_calib = self.oseries.series.loc[tmin:tmax]
# sample measurements, so that frequency is not higher than model
# keep the original timestamps, as they will be used during
# interpolation of the simulation
sim_index = self.get_sim_index(tmin, tmax, freq,
self.settings["warmup"])
if not oseries_calib.empty:
index = get_sample(oseries_calib.index, sim_index)
oseries_calib = oseries_calib.loc[index]
if not update_observations:
# tmin, tmax and freq are equal to the settings
# so we can set self.oseries_calib to improve speed of next run
self.oseries_calib = oseries_calib
else:
oseries_calib = self.oseries_calib
return oseries_calib