Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Improved ComEd Hourly Pricing python script
#!/usr/bin/env python2
import argparse
import locale
import logging
import time
import threading
import json
import requests
import datetime
#from aiy.board import Board
#from aiy.voice.audio import AudioFormat, play_wav, record_file, Recorder
#from aiy.board import Board, Led
#from aiy.cloudspeech import CloudSpeechClient
# logging levels
# - FATAL
# - ERROR
# - WARNING
# - INFO
# - DEBUG
# - NOTSET
def main():
COMED_HP_API_BASE_URL = 'https://hourlypricing.comed.com/api?type='
parser = argparse.ArgumentParser(description='ComEd Hourly Pricing API Data Access')
parser.add_argument('--pricetype', '-t', default='hour')
parser.add_argument('--datapoints', '-n', type=int, default=1, required=False)
parser.add_argument('--verbosity', '-v', default='none', required=False)
parser.add_argument('--format', '-f', default='plain', required=False)
args = parser.parse_args()
if args.verbosity == 'debug':
logging.basicConfig(level=logging.DEBUG)
elif args.verbosity == 'info':
logging.basicConfig(level=logging.INFO)
elif args.verbosity == 'warning':
logging.basicConfig(level=logging.WARN)
elif args.verbosity == 'error':
logging.basicConfig(level=logging.ERROR)
else:
logging.basicConfig(level=logging.FATAL)
logging.debug('The --pricetype argument provided was: %s', args.pricetype)
if args.pricetype == 'hour':
comed_hp_api_url = COMED_HP_API_BASE_URL + 'currenthouraverage'
logging.info('Getting the current hour average price from the ComEd Hourly Pricing API...')
elif args.pricetype == '5minute':
comed_hp_api_url = COMED_HP_API_BASE_URL + '5minutefeed'
logging.info('Getting the 5-minute price data from the ComEd Hourly Pricing API...')
logging.debug('You requested %d datapoints from the 5-minute price feed', args.datapoints)
# Get JSON data, and store it in a list
response = requests.get(comed_hp_api_url)
pricedatalist = json.loads(response.text)
logging.info('Price data has been retrieved, and stored as JSON data.')
logging.debug("Raw Price Data:\n\t\t%s", pricedatalist)
if args.datapoints == 1:
# select the first datapoint from the price data list
pricedata = pricedatalist[0]
# store the price time as an integer (in milliseconds since the UNIX epoch)
mtimeupdated = int(pricedata.get("millisUTC"))
# store the current price, in cents, as a floating-point value
currentprice = float(pricedata.get("price"))
# create a useable time from the integer millis value
timeupdated = datetime.datetime.fromtimestamp(mtimeupdated / 1000.0)
# store the current price as a portion of a dollar
normalized_price = currentprice / 100.0
logging.info("The price data was last updated at %s", timeupdated)
logging.info("The price as of %s was %s cents per kWh", timeupdated, currentprice)
print("The price as of %s was: $%.3f/kWh" % (timeupdated, (currentprice / 100.0)))
else:
current_datapoint = 0
if args.format == 'csv':
print("sequence,datetime,price")
while current_datapoint < args.datapoints:
logging.debug("Currently processing datapoint %d of %d total", current_datapoint, args.datapoints)
# process the current datapoint
dp_pricedata = pricedatalist[current_datapoint]
dp_millitime = int(dp_pricedata.get("millisUTC"))
dp_timeprice = float(dp_pricedata.get("price"))
dp_humantime = datetime.datetime.fromtimestamp(dp_millitime / 1000.0)
dp_pricefull = dp_timeprice / 100.0
if args.format == 'csv':
print("%d,%s,%.3f" % ((current_datapoint + 1), dp_humantime, dp_pricefull))
else:
logging.info("Datapoint processed")
print("[%d] The price at %s was: $%.3f/kWh" % ((current_datapoint + 1), dp_humantime, dp_pricefull))
logging.debug("Incrementing datapoint...")
current_datapoint = current_datapoint + 1
logging.debug("The new datapoint is %d", current_datapoint)
# for x in pricedata.values():
# print(x)
if __name__ == '__main__':
main()
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.