Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_anomaly_calc_given_climatology_return_clim():
clim = pd.Series(np.arange(366), name='clim', index=np.arange(366) + 1)
data = pd.Series(
np.arange(366), index=pd.date_range('2000-01-01', periods=366))
anom_should = pd.Series(
np.zeros(366), index=pd.date_range('2000-01-01', periods=366),
name='anomaly')
anom = anomaly.calc_anomaly(
data, climatology=clim, respect_leap_years=False,
return_clim=True)
pdt.assert_series_equal(anom_should, anom['anomaly'], check_dtype=False)
def test_anomaly_calc_given_climatology_no_leap_year():
clim = pd.Series(np.arange(366), name='clim', index=np.arange(366) + 1)
data = pd.Series(
np.arange(365), index=pd.date_range('2007-01-01', periods=365))
anom_should = pd.Series(
np.zeros(365), index=pd.date_range('2007-01-01', periods=365))
anom = anomaly.calc_anomaly(
data, climatology=clim, respect_leap_years=True)
pdt.assert_series_equal(anom_should, anom, check_dtype=False)
def calc_anom(self, data):
if self.columns is None:
ite = data
else:
ite = self.columns
for column in ite:
clim = calc_climatology(data[column], **self.kwargs)
data[column] = calc_anomaly(data[column], climatology=clim)
return data
axes.append(ax)
ax.xaxis.set_visible(False)
axes.append(last_axis)
else:
own_axis = False
for i, column in enumerate(df):
Ser = df[column]
ax = axes[i]
if clim is None:
clima = anom.calc_climatology(Ser)
else:
clima = pd.Series(clim[column])
anomaly = anom.calc_anomaly(Ser, climatology=clima, return_clim=True)
anomaly[Ser.name] = Ser
anomaly = anomaly.dropna()
pos_anom = anomaly[Ser.name].values > anomaly['climatology'].values
neg_anom = anomaly[Ser.name].values < anomaly['climatology'].values
ax.plot(anomaly.index, anomaly[Ser.name].values, 'o',
markersize=markersize, mfc=mfc, mec=mec)
ax.plot(anomaly.index, anomaly['climatology'].values,
linestyle=clim_linestyle,
color=clim_color,
linewidth=clim_linewidth)
ax.fill_between(anomaly.index,
def calc_anom(self, data):
if self.columns is None:
ite = data
else:
ite = self.columns
for column in ite:
data[column] = calc_anomaly(data[column],
window_size=self.window_size)
return data
#
#calculate anomaly based on moving +- 17 day window
anomaly = ts.anomaly.calc_anomaly(ascat_ts.data['sm'], window_size=35)
anomaly.plot()
#
#calculate climatology
climatology = ts.anomaly.calc_climatology(ascat_ts.data['sm'])
climatology.plot()
#
#calculate anomaly based on climatology
anomaly_clim = ts.anomaly.calc_anomaly(ascat_ts.data['sm'], climatology=climatology)
anomaly_clim.plot()
'ASCAT_WARP5.5_R1.2','080_ssm','netcdf')
ascat_grid_folder = os.path.join('R:\\','Datapool_processed','WARP','ancillary','warp5_grid')
#init the ASCAT_SSM reader with the paths
ascat_SSM_reader = ascat.AscatH25_SSM(ascat_folder,ascat_grid_folder)
#
ascat_ts = ascat_SSM_reader.read_ssm(45,0)
#plot soil moisture
ascat_ts.data['sm'].plot()
#
#calculate anomaly based on moving +- 17 day window
anomaly = ts.anomaly.calc_anomaly(ascat_ts.data['sm'], window_size=35)
anomaly.plot()
#
#calculate climatology
climatology = ts.anomaly.calc_climatology(ascat_ts.data['sm'])
climatology.plot()
#
#calculate anomaly based on climatology
anomaly_clim = ts.anomaly.calc_anomaly(ascat_ts.data['sm'], climatology=climatology)
anomaly_clim.plot()