In stefwayland/cecmidas: Interact with the California Energy Commission MIDAS Rate Database API

library(reticulate)
use_condaenv("base")

MIDAS Demo Code

Variable setup and base64 encoding

import requests
import json
import base64

# base_url = "https://midasapi.energy.ca.gov/api"
base_url = "http://accmidasapi/api"

##Encode registration values as base64

#organization is an optional value in the registration API
organization = "CEC"
org_encodedBytes = base64.b64encode(organization.encode("utf-8"))
organization64 = str(org_encodedBytes, "utf-8")

username = "stest"
user_encodedBytes = base64.b64encode(username.encode("utf-8"))
username64 = str(user_encodedBytes, "utf-8")

password = "Pantseveninghero1-"
pswd_encodedBytes = base64.b64encode(password.encode("utf-8"))
password64 = str(pswd_encodedBytes, "utf-8")

emailaddress = "[e-mail]"
email_encodedBytes = base64.b64encode(emailaddress.encode("utf-8"))
emailaddress64 = str(email_encodedBytes, "utf-8")

fullname = "Stefanie Wayland"
fullname_encodedBytes = base64.b64encode(fullname.encode("utf-8"))
fullname64 = str(fullname_encodedBytes, "utf-8")

Registration

#Put together the dict for the JSON payload
registration_info = {"organization":organization64,"username":username64,"password":password64,"emailaddress":emailaddress64,"fullname":fullname64}

url = base_url + '/registration'
headers =  {"Content-Type":"application/json"}

response = requests.post(url, data=json.dumps(registration_info), headers=headers)

#Prints below will return 200 response for successful call
print(response)
#Response text should be: 'User account for <your_user_name> was successfully created. A verification email has been sent to <your_email>. Please click the link in the email in order to start using the API.'
print(response.text)

Get Token

credentials = username + ":" + password
credentials_encodedBytes = base64.b64encode(credentials.encode("utf-8"))

headers = {b'Authorization': b'BASIC ' + credentials_encodedBytes}
url = base_url + '/token'

response = requests.get(url, headers = headers)
print(response.text)

token = response.headers['Token']
print(token)

Get RIN List

# parameter signaltype acceptable values:
# 0 = All signal types
# 1 = Tariff signals only
# 2 = Green House Gas Emissions only
# 3 = Flex Alerts only

signaltype = '1'

headers = {'accept': 'application/json', 'Authorization': "Bearer " + token}
url = base_url + '/valuedata?signaltype=' + signaltype
list_response = requests.get(url, headers = headers)
print(json.dumps(json.loads(list_response.text), indent = 2))

Get XML Schema

import xml.etree.ElementTree as ET

headers = {'accept': 'application/json', 'Authorization': "Bearer " + token}
url = base_url + '/valuedata'
pricing_response = requests.get(url, headers = headers)

element = ET.XML(json.loads(pricing_response.text))
ET.indent(element)
print(ET.tostring(element, encoding='unicode'))

Get Values

queryType = 'alldata'
# rateID = 'USCA-TSTS-HTOU-TEST'
rateID = 'USCA-TSTS-HTOU-0000'
headers = {'accept': 'application/json', 'Authorization': "Bearer " + token}
url = base_url + '/valuedata?id=' + rateID + '&querytype=' + queryType
pricing_response = requests.get(url, headers = headers)
print(json.dumps(json.loads(pricing_response.text), indent = 2))
print(pricing_response)

Upload Rates

import os 
import sys 

# some debugging help
import http.client
# logging stuff
http.client.HTTPConnection.debuglevel = 1

# File on the local filesystem that is correctly formatted against the XML schema definition (XSD)
# priceFileName = 'Streaming_Test.xml'
priceFileName = 'MIDAS_Test_Rate_24TOU.xml'

headers = {'accept': 'application/json', 'Content-Type': 'text/xml', 'Authorization': "Bearer " + token}
url = base_url + '/valuedata'
priceFile = open(priceFileName)
xml = priceFile.read()
priceFile.close()
pricing_response = requests.post(url, data = xml, headers = headers)

# print(pricing_response.request.body)
print(pricing_response.text)
print(pricing_response)

Get Lookup Table Data

LookupTable = 'Location'

headers = {'accept': 'application/json', 'Authorization': "Bearer " + token}
url = base_url + '/valuedata?' + 'LookupTable=' + LookupTable
pricing_response = requests.get(url, headers = headers)
response = requests.get(url, headers = headers)
print(json.dumps(json.loads(pricing_response.text), indent = 2))

Get Holiday Table Data

headers = {'accept': 'application/json', 'Authorization': "Bearer " + token}
url = base_url + '/holiday'
pricing_response = requests.get(url, headers=headers)
response = requests.get(url,headers=headers)
print(json.dumps(json.loads(pricing_response.text), indent = 2))

Upload Holidays

import os 
import sys 

# Make certain file on the local filesystem is correctly formatted against the XML schema definition (XSD)
holidayFileName = 'MIDAS_Test_Holidays.xml'

headers = {'accept': 'application/json', 'Content-Type': 'text/xml', 'Authorization': "Bearer " + token}
url = base_url + '/holiday'
holidayFile = open(holidayFileName)
xml = holidayFile.read()
holidayFile.close()
print(xml)
holiday_response = requests.post(url, data = xml, headers = headers)

print(holiday_response.text)
print(holiday_response)

Get List of RINs with Historical Data

DistributionCode = 'TS'
EnergyCode = 'TS'

headers = {'accept': 'application/json', 'Authorization': "Bearer " + token}
url = base_url + '/historicallist?' + 'DistributionCode=' + DistributionCode + '&EnergyCode=' + EnergyCode
list_response = requests.get(url, headers = headers)
print(json.dumps(json.loads(list_response.text), indent = 2))

Get Historical Data

rateID = 'USCA-TSTS-HTOU-0000'
startdate = '2022-06-20'
enddate = '2022-12-25'

headers = {'accept': 'application/json', 'Authorization': "Bearer " + token}
url = base_url + '/historicaldata?id=' + rateID + '&startdate=' + startdate + '&enddate=' + enddate
list_response = requests.get(url, headers=headers)
print(json.dumps(json.loads(list_response.text), indent = 2))

stefwayland/cecmidas documentation built on Aug. 27, 2023, 2:36 p.m.