Create a Dispatch Polling Integration
In this tutorial, we will describe the basics of polling for dispatches from our system. This approach does not require a public-facing server on the Internet but requires you to regularly contact Voltus for updates. Where possible, we recommend using webhooks instead.
The basics of this approach are pretty simple: contact the dispatches endpoint regularly using your API key, and see if there are any upcoming or currently active dispatches. If there are, take the appropriate action.
The initial steps of this tutorial can be accomplished without an API Key (using test credentials).
For the last two steps ("Running a Sandbox Dispatch Simulation" and "Connecting to Production") you’ll create test dispatches and access them with your Sandbox (SANDBOX_API_KEY) and Production (VOLTUS_API_KEY) API keys, respectively.
If you do not have API keys, reach out to your Voltus point of contact or api-support@voltus.co.
While we provide source code examples in-line for illustrative purposes, we recommend checking out the source code from GitHub in the voltus-api-examples repository if you want to follow along interactively. Run the initial setup guidelines outlined in the repository’s README.md. Some parts of this tutorial also assume you have a small utility called curl installed.
Retrieving dispatch data
The first step is to just retrieve the data for the dispatches endpoint. Let’s test out how this might work, using testing credentials.
import json
import os
import urllib
import requests
VOLTUS_SANDBOX_API_URL = "https://sandbox.voltus.co"
VOLTUS_TEST_API_KEY = "secret"
s = requests.Session()
s.headers.update({"X-Voltus-API-Key": VOLTUS_TEST_API_KEY, "Accept": "application/json"}
)
url = urllib.parse.urljoin(VOLTUS_API_URL, "/2022-04-15/dispatches")
r = s.get(url)
print(json.dumps(r.json(), indent=4))
If you run this program, you should get something like this as output:
{
"dispatches": [
{
"id": "asdf",
"authorized": true,
"test": false,
"start_time": "2023-06-29T14:53:36.563070478Z",
"end_time": "2023-06-29T15:53:36.563069928Z",
"program": {
"name": "MISO - Operating Reserves",
"timezone": "US/Central",
"market": "MISO",
"program_type": "ancillary_services"
},
"sites": [
{
"name": "Site 1",
"id": "asdf",
"customer_location_id": "1",
"commitment": 12
},
{
"name": "Site 2",
"id": "lkjh",
"customer_location_id": "2",
"commitment": 40
}
]
}
],
"page": 0,
"perPage": 0
}
In the example above, we received a dispatches object with one single dispatch in it.
The key fields to pay attention to above are authorized, test, start_time, end_time, and sites.
authorizedindicates whether the dispatch is authorized or not. If authorized is ever set to false, the dispatch is considered canceled and you should not curtail.testindicates whether a dispatch is a test or not. This is to distinguish between test dispatches and dispatches scheduled by Voltus in response to a market dispatch. Unless testing the former, you can ignore any dispatch wheretestis true.start_timeandend_timeindicate when the dispatch starts and ends, respectively.
Processing dispatch data
Let’s use what we learned above and write a program that checks to see if a dispatch is ongoing and sends a dispatch signal to specific sites if that is the case.
This source code to this program can be found in voltus-api-examples as polling/processor.py.
"""
Simplified polling processor which gets dispatches then waits to see if they're in progress
Precursor to the more sophisticated poller.py
"""
import datetime
import os
import time
import requests
import urllib
from dateutil import parser as date_parser
VOLTUS_API_URL = os.getenv("VOLTUS_API_URL", "https://sandbox.voltus.co")
VOLTUS_API_KEY = os.getenv("VOLTUS_API_KEY", "secret")
processed_dispatches = set()
s = requests.Session()
s.headers.update({"X-Voltus-API-Key": VOLTUS_API_KEY, "Accept": "application/json"})
url = urllib.parse.urljoin(VOLTUS_API_URL, "/2022-04-15/dispatches")
resp = s.get(url)
dispatches = resp.json()["dispatches"]
if __name__ == "__main__":
while True:
for dispatch_info in dispatches:
if dispatch_info["id"] in processed_dispatches:
continue
start_time = date_parser.parse(dispatch_info["start_time"])
end_time = (
date_parser.parse(dispatch_info["end_time"])
if dispatch_info.get("end_time")
else None
)
now = datetime.datetime.now(datetime.timezone.utc)
if start_time <= now:
if end_time is None or end_time >= now:
processed_dispatches.add(dispatch_info["id"])
print("Dispatch {} is in progress".format(dispatch_info["id"]))
if dispatch_info["authorized"]:
for site in dispatch_info["sites"]:
print(
"- Customer location {} should curtail. Commitment: {} kW".format(
site["customer_location_id"], site["commitment"]
)
)
else:
print("- Dispatch is not authorized or is cancelled, skipping")
else:
print("Dispatch {} has ended".format(dispatch_info["id"]))
time.sleep(1)
If you run the above, you should get some output like the following after a very short delay:
Dispatch asdf is in progress
- Customer location 1 should curtail. Commitment: 12 kW
- Customer location 2 should curtail. Commitment: 40 kW
Putting it together
We can build on the approach above and create a program, which continuously scans for new dispatches in a loop and sends updates where appropriate. We’ll design this as a skeleton and leave it as an exercise for the reader to fill in the logic to respond to dispatches.
This source code to this program can be found in voltus-api-examples as polling/poller.py.
"""
Skeleton program which continually polls for new dispatches
"""
import os
import urllib
import datetime
import time
from typing import Dict
import requests
from dateutil import parser as date_parser
VOLTUS_API_URL = os.getenv("VOLTUS_API_URL", "https://sandbox.voltus.co")
VOLTUS_API_KEY = os.getenv("VOLTUS_API_KEY", "secret")
POLLING_INTERVAL_SECONDS = 30
s = requests.Session()
s.headers.update({"X-Voltus-API-Key": VOLTUS_API_KEY, "Accept": "application/json"})
managed_dispatches: Dict[str, Dict] = {} # dispatch id -> dispatch info
cancelled_dispatches = set()
def create_dispatch(dispatch_info: Dict):
# TODO: Fill in with your company-specific logic
# Remember that this could get called twice if the service gets restarted
# Remember that end_time could be None
# sites could have one sites, many sites, or all of your sites
print(
"Dispatch with ID {} starting at {} and ending at {} for sites {} created".format(
dispatch_info["id"],
dispatch_info["start_time"],
dispatch_info.get("end_time", "not specified"),
dispatch_info["sites"],
)
)
def update_dispatch(dispatch_info: Dict):
# TODO: Fill in with your company-specific logic
# start_time and end_time can change to be earlier or later
# Remember that this could get called twice if the service is restarted or the request otherwise fails
print(
"Dispatch with ID {} starting at {} and ending at {} for sites {} updated".format(
dispatch_info["id"],
dispatch_info["start_time"],
dispatch_info.get("end_time", "not specified"),
dispatch_info["sites"],
)
)
def cancel_dispatch(dispatch_info: Dict):
# TODO: Fill in with your company-specific logic to cancel this dispatch, likely by ID
# Remember that this could get called twice if the service is restarted
print("Dispatch with ID {} cancelled".format(dispatch_info["id"]))
if __name__ == "__main__":
while True:
start_time = datetime.datetime.now(datetime.timezone.utc)
url = urllib.parse.urljoin(VOLTUS_API_URL, "/2022-04-15/dispatches")
resp = s.get(url)
resp.raise_for_status() # raise an exception if request fails
dispatches = resp.json()
for dispatch_info in dispatches["dispatches"]:
# convert start/end time to proper datetime
dispatch_info["start_time"] = date_parser.parse(dispatch_info["start_time"])
if "end_time" in dispatch_info and dispatch_info["end_time"] is not None:
dispatch_info["end_time"] = date_parser.parse(dispatch_info["end_time"])
else:
dispatch_info["end_time"] = None
dispatch_id = dispatch_info["id"]
now = datetime.datetime.now(datetime.timezone.utc)
# if we've already cancelled this dispatch, skip it
if dispatch_id in cancelled_dispatches:
continue
# send a cancelled message if the dispatch is over or not authorized
# regardless of whether we know about it or not (to protect against
# this service restarting or similar), as long as it has not been
# previously marked as cancelled
if (
dispatch_info.get("end_time") and dispatch_info["end_time"] < now
) or not dispatch_info["authorized"]:
cancel_dispatch(dispatch_info)
cancelled_dispatches.add(dispatch_id)
managed_dispatches.pop(dispatch_id, None)
# if this dispatch hasn't been seen before, create it
elif dispatch_id not in managed_dispatches:
create_dispatch(dispatch_info)
managed_dispatches[dispatch_id] = dispatch_info
# otherwise, update it
else:
if managed_dispatches[dispatch_id] != dispatch_info:
update_dispatch(dispatch_info)
managed_dispatches[dispatch_id].update(dispatch_info)
# sleep for 30 seconds minus however long it took to process the dispatches
elapsed = datetime.datetime.now(datetime.timezone.utc) - start_time
time.sleep(max(0, POLLING_INTERVAL_SECONDS - elapsed.total_seconds()))
If you run this using the test credentials (the default), you’ll get an initial message saying that it created a dispatch:
Dispatch with ID asdf starting at 2023-07-31 16:38:59.739840+00:00 and ending at 2023-07-31 17:38:59.739840+00:00 for sites [{'name': 'Site 1', 'id': 'asdf', 'customer_location_id': '1', 'commitment': 12}, {'name': 'Site 2', 'id': 'lkjh', 'customer_location_id': '2', 'commitment': 40}] created
Then a stream of updates that looks like this:
Dispatch with ID asdf starting at 2023-07-31 16:39:29.467542+00:00 and ending at 2023-07-31 17:39:29.467541+00:00 for sites [{'name': 'Site 1', 'id': 'asdf', 'customer_location_id': '1', 'commitment': 12}, {'name': 'Site 2', 'id': 'lkjh', 'customer_location_id': '2', 'commitment': 40}] updated
This is because the public sandbox API key always returns a single dispatch, scheduled a short period in the future.
Let’s create a more realistic scenario by using our sandbox API key to create a dispatch scenario, then watch for updates with a properly configured version of the above program.
Running a Sandbox Dispatch Simulation
We will create a basic dispatch simulation in sandbox.
Change the VOLTUS_API_KEY to the Sandbox API Key voltus sent in code or by setting the environment variable with:
On Linux or MacOS:
export VOLTUS_API_URL = "https://sandbox.voltus.co"
export VOLTUS_API_KEY = "<SANDBOX_API_KEY>"
Or on Windows Powershell:
$Env:VOLTUS_API_URL = "https://sandbox.voltus.co"
$Env:VOLTUS_API_KEY = "<SANDBOX_API_KEY>"
Run the dispatch scenario by running the curl command with a near-future notification_time at Sandbox Dispatch Simulations - Basic Dispatch.
Before running, you'll need to change the notification_time, start_time, and end_time to now or some time in the near future.
Now, if we re-run the program above, we should get something like the following as output:
Dispatch with ID r4y5 starting at 2024-12-01T16:50:00+00:00 and ending at 2024-12-01T17:00:00+00:00 for sites [{'name': 'My Site', 'id': 'd4z9', 'customer_location_id': 'democustomerlocationid', 'drop_by': 5},{'name': 'Another Site', 'id': '9jeq', 'customer_location_id': 'democustomerlocationid2', 'drop_by': 50},] created
Next, run each dispatch simulations outlined in Sandbox Dispatch Simulations and confirm your software works as expected for each scenario.
Connecting to Production
Once you've confirmed your software works as expected with the sandbox dispatch simulations, test your connection to the production API. Set the VOLTUS_API_KEY environment variable to your production API key.
On Linux or MacOS:
export VOLTUS_API_URL = "https://api.voltus.co"
export VOLTUS_API_KEY = "<VOLTUS_API_KEY>"
Or on Windows Powershell:
$Env:VOLTUS_API_URL = "https://api.voltus.co"
$Env:VOLTUS_API_KEY = "<VOLTUS_API_KEY>"
This will make the service contact the production version of the Voltus API. Now, create a test dispatch by running the following via curl (set the start and end times to sometime in the future before doing so, otherwise the request will fail):
curl --request POST \
--url https://api.voltus.co/2022-04-15/dispatches \
--header 'Content-Type: application/json' \
--header "X-Voltus-API-Key: ${VOLTUS_API_KEY}" \
--data '{
"start_time": "2023-07-06T19:30:00-07:00",
"end_time": "2023-07-06T19:40:00-07:00"
}'
Now, if we re-run the program above, we should get something like the following as output (depending on your exact configuration with Voltus):
Dispatch with ID r4y5 starting at 2023-07-31 16:38:59+00:00 and ending at 2023-07-31 17:38:59+00:00 for sites [{'name': 'Test Site 1', 'id': 'd4z9', 'customer_location_id': '1', 'drop_by': 5}] created
Next steps
Congratulations! You have successfully set up and tested a basic dispatch service.
After passing all the simulated dispatches in sandbox and confirming your connection to production with a self-scheduled test dispatch, your integration is ready to accept new sites. Contact your Sales Director and api-support@voltus.co for next steps. Each new site must still pass a dispatch verification test. Dispatch verifications are site-specific end-to-end tests in the production environment.