Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Solar Forecast Arbiter Test Trial 1 Script
"""
An example script to generate random forecasts for participants
in the first test trial of the Solar Forecast Arbiter.
This script should be run as cronjob or via another scheduling
mechanism at an appropriate interval that will be determined by
the trial/forecast parameters.
"""
import logging
import sys
import numpy as np
import pandas as pd
from solarforecastarbiter.io import api
from solarforecastarbiter.reference_forecasts import utils
TRIAL_NAME = 'Test Trial 1'
API_URL = 'https://api.solarforecastarbiter.org'
logging.basicConfig(level='INFO')
# here, we read the file provided as an argument to the script
# to get the username and password (separated by a new line).
# Alternatives include using environment variables or hardcoding the values
with open(sys.argv[1], 'r') as f:
username, password = f.read().split('\n')[:2]
# Setup an APISession to communicate with the solararbiter API
token = api.request_cli_access_token(username, password)
session = api.APISession(token, base_url=API_URL)
# Get information about the current user
user_info = session.get_user_info()
# Retrive all forecasts the user has access to
all_forecasts = session.list_forecasts()
# Filter out the forecasts not in the trial
# for the purposes of this trial, the trial name will
# appear in the extra_parameters section of the Forecast
trial_forecasts = filter(
lambda x: (
TRIAL_NAME in x.extra_parameters
) and (
x.provider == user_info['organization']
),
all_forecasts
)
# go through each of our forecasts in the trial,
# generate random data, and upload to the API
for forecast in trial_forecasts:
logging.info('Check if a forecast should be generated for %s',
forecast.name)
# set the run_time as now
run_time = pd.Timestamp.now(tz='UTC')
# From the forecast metadata, determine the next time
# the forecasts should be issued
issue_time = utils.get_next_issue_time(
forecast, run_time)
# if the next issue_time is not within 10 minutes of the
# current time, skip and move on to the next forecast
if (issue_time - run_time) > pd.Timedelta('10min'):
logging.info('Not yet time to generate forecast for %s',
forecast.name)
continue
# Get the time range that we are expected to generate a
# forecast for. This includes an adjustment for the lead time
# before a forecast is valid.
start, end = utils.get_forecast_start_end(
forecast, issue_time, adjust_for_interval_label=True)
logging.info('Generating forecast for %s from %s to %s',
forecast.name, start, end)
# make the forecast, in this case just random numbers
# between 0 and 100
# first, make the index
index = pd.date_range(start=start, end=end, freq=forecast.interval_length)
# now make the random series
forecast_series = pd.Series(np.random.randint(0, 100, len(index)),
index=index)
# upload the forecast to the API
# catch and log errors so we can try uplloading the other forecasts
try:
session.post_forecast_values(forecast.forecast_id, forecast_series)
except Exception:
logging.exception('Failed to upload forecast for %s', forecast.name)
continue
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.