Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
forecast : pd.Series
The persistence forecast. The forecast interval label is the
same as the observation interval label.
"""
# ensure that we're using times rounded to multiple of interval_length
_check_intervals_times(observation.interval_label, data_start, data_end,
forecast_start, forecast_end,
observation.interval_length)
# get observation data for specified range
obs = load_data(observation, data_start, data_end)
# partial-up the metadata for solar position and
# clearsky calculation clarity and consistency
site = observation.site
calc_solpos = partial(pvmodel.calculate_solar_position,
site.latitude, site.longitude, site.elevation)
calc_cs = partial(pvmodel.calculate_clearsky,
site.latitude, site.longitude, site.elevation)
# Calculate solar position and clearsky for obs time range.
# if data is instantaneous, calculate at the obs time.
# else (if data is interval average), calculate at 1 minute resolution to
# reduce errors from changing solar position during persistence data range.
# Later, modeled clear sky or ac power will be averaged over the data range
closed = datamodel.CLOSED_MAPPING[observation.interval_label]
if closed is None:
freq = observation.interval_length
else:
freq = pd.Timedelta('1min')
obs_range = pd.date_range(start=data_start, end=data_end, freq=freq,
closed=closed)
def _solpos_night(observation, values):
solar_position = pvmodel.calculate_solar_position(
observation.site.latitude, observation.site.longitude,
observation.site.elevation, values.index)
night_flag = validator.check_irradiance_day_night(solar_position['zenith'],
_return_mask=True)
return solar_position, night_flag
# instantaneous values.
freq = '5min'
start_adj, end_adj = adjust_start_end_for_interval_label(interval_label,
start, end)
cloud_cover = forecast.reindex_fill_slice(
cloud_cover, freq=freq, start=start, end=end,
start_slice=start_adj, end_slice=end_adj,
fill_method=fill_method)
resample_fill_slicer = partial(
forecast.reindex_fill_slice, freq=freq, start=start, end=end,
start_slice=start_adj, end_slice=end_adj, fill_method='interpolate')
air_temperature, wind_speed = [
resample_fill_slicer(v) for v in (air_temperature, wind_speed)
]
if solar_position is None:
solar_position = pvmodel.calculate_solar_position(
latitude, longitude, elevation, cloud_cover.index)
ghi, dni, dhi = forecast.cloud_cover_to_irradiance(
latitude, longitude, elevation, cloud_cover,
solar_position['apparent_zenith'], solar_position['zenith'])
label = datamodel.CLOSED_MAPPING[interval_label]
resampler = partial(forecast.resample, freq='1h', label=label)
def solar_pos_calculator(): return solar_position
return (ghi, dni, dhi, air_temperature, wind_speed,
resampler, solar_pos_calculator)
def _ghi_to_dni_dhi(latitude, longitude, elevation, ghi):
"""
Calculate DNI, DHI from GHI and calculated solar position.
"""
solar_position = pvmodel.calculate_solar_position(
latitude, longitude, elevation, ghi.index)
dni, dhi = pvmodel.complete_irradiance_components(
ghi, solar_position['zenith'])
def solar_pos_calculator(): return solar_position
return dni, dhi, solar_pos_calculator