summaryrefslogtreecommitdiff
path: root/app
diff options
context:
space:
mode:
Diffstat (limited to 'app')
-rw-r--r--app/constants/constants.py22
-rw-r--r--app/energy_service.py20
-rw-r--r--app/request/energy/energy_request.py118
-rw-r--r--app/request/login/login_request.py60
4 files changed, 110 insertions, 110 deletions
diff --git a/app/constants/constants.py b/app/constants/constants.py
index f532a9d..bbfe97b 100644
--- a/app/constants/constants.py
+++ b/app/constants/constants.py
@@ -2,7 +2,7 @@ from app.request.login.credentials import *
from enum import Enum
# Useless response key/pairs
-uselessResponseKeys = ['DemandColorCode', 'UsageDelvcolorCode', 'UsageReccolorCode',
+useless_response_keys = ['DemandColorCode', 'UsageDelvcolorCode', 'UsageReccolorCode',
'UsageColorCode', 'SoFarColorCode', 'ExpectedUsageColorCode',
'PeakLoadColorCode', 'AverageColorCode', 'LoadFactorColorCode',
'HighestColorCode', 'IsOnlyAMI', 'Skey', 'AccountNumber',
@@ -17,7 +17,7 @@ class RequestMode(Enum):
month = 'M'
# Shared
-genericRequestHeaders = {
+generic_request_headers = {
# set csrftoken
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'en-US,en;q=0.9,la;q=0.8',
@@ -40,7 +40,7 @@ genericRequestHeaders = {
}
################################################################
# Login
-loginPageHeaders = {
+login_page_headers = {
# no setting needed
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Accept-Language': 'en-US,en;q=0.9,la;q=0.8',
@@ -59,7 +59,7 @@ loginPageHeaders = {
'sec-ch-ua-platform': '"Linux"',
}
-loginRequestJson = {
+login_request_json = {
# no setting needed
'username': username,
'password': password,
@@ -72,7 +72,7 @@ loginRequestJson = {
################################################################
# Usage
-usageRequestCookies = {
+usage_request_cookies = {
# set the following:
# ApplicationGatewayAffinityCORS
# ApplicationGatewayAffinity
@@ -87,7 +87,7 @@ usageRequestCookies = {
'UName': uName,
}
-electricUsageRequestJson = {
+electric_usage_request_json = {
# Set Mode
'UsageOrGeneration': '1',
'Type': 'K',
@@ -101,7 +101,7 @@ electricUsageRequestJson = {
'DateToDaily': '',
}
-waterUsageRequestJson = {
+water_usage_request_json = {
# Set Mode
"Type":"W",
"strDate": "",
@@ -118,7 +118,7 @@ waterUsageRequestJson = {
################################################################
# Endpoints/URIs
-waterRequestEndpoint = 'https://myaccount.cityutilities.net/Portal/Usages.aspx/LoadWaterUsage'
-electricRequestEndpoint = 'https://myaccount.cityutilities.net/Portal/Usages.aspx/LoadUsage'
-loginRequestEndpoint = 'https://myaccount.cityutilities.net/Portal/Default.aspx/validateLogin'
-loginPageUri = 'https://myaccount.cityutilities.net/Portal/default.aspx' \ No newline at end of file
+water_request_endpoint = 'https://myaccount.cityutilities.net/Portal/Usages.aspx/LoadWaterUsage'
+electric_request_endpoint = 'https://myaccount.cityutilities.net/Portal/Usages.aspx/LoadUsage'
+login_request_endpoint = 'https://myaccount.cityutilities.net/Portal/Default.aspx/validateLogin'
+login_page_uri = 'https://myaccount.cityutilities.net/Portal/default.aspx'
diff --git a/app/energy_service.py b/app/energy_service.py
index 2eb6867..05ed11b 100644
--- a/app/energy_service.py
+++ b/app/energy_service.py
@@ -1,23 +1,23 @@
from fastapi import FastAPI
from app.request.login.login_request import login
-from app.request.energy.energy_request import requestUsageData
+from app.request.energy.energy_request import request_usage_data
import os
import json
app = FastAPI(docs_url=None, redoc_url=None)
-debugFilePath = "./get_response.json"
+debug_file_path = "./get_response.json"
@app.get('/get')
def get():
- if (debugFileExists()):
+ if (debug()):
print('Debug file present')
- return getDebugResponse()
- sessionKeys = login()
- return requestUsageData(sessionKeys)
+ return get_debug_response()
+ session_keys = login()
+ return request_usage_data(session_keys)
-def getDebugResponse():
- with open(debugFilePath) as file:
+def get_debug_response():
+ with open(debug_file_path) as file:
return json.load(file)
-def debugFileExists():
- return os.path.isfile(debugFilePath) \ No newline at end of file
+def debug():
+ return os.path.isfile(debug_file_path)
diff --git a/app/request/energy/energy_request.py b/app/request/energy/energy_request.py
index 20c65d3..15d6322 100644
--- a/app/request/energy/energy_request.py
+++ b/app/request/energy/energy_request.py
@@ -1,103 +1,103 @@
import json
import requests
-from ...constants.constants import usageRequestCookies, genericRequestHeaders, electricUsageRequestJson, waterUsageRequestJson, waterRequestEndpoint, electricRequestEndpoint, RequestMode, uselessResponseKeys
+from ...constants.constants import usage_request_cookies, generic_request_headers, electric_usage_request_json, water_usage_request_json, water_request_endpoint, electric_request_endpoint, RequestMode, useless_response_keys
from datetime import date
from datetime import timedelta
# Electric
-def doElectricRequest(requestMode: RequestMode):
- electricUsageRequestJson['Mode'] = requestMode.value
- if (requestMode in [RequestMode.halfHour, RequestMode.hour]):
- electricUsageRequestJson['strDate'] = getYesterday()
- return performElectricRequest()
+def do_electric_request(request_mode: RequestMode):
+ electric_usage_request_json['Mode'] = request_mode.value
+ if (request_mode in [RequestMode.halfHour, RequestMode.hour]):
+ electric_usage_request_json['strDate'] = get_yesterday_date()
+ return perform_electric_request()
-def performElectricRequest():
- electricUsageResponse = requests.post(
- electricRequestEndpoint,
- cookies=usageRequestCookies,
- headers=genericRequestHeaders,
- json=electricUsageRequestJson
+def perform_electric_request():
+ electric_usage_response = requests.post(
+ electric_request_endpoint,
+ cookies=usage_request_cookies,
+ headers=generic_request_headers,
+ json=electric_usage_request_json
)
- return parseResponse(electricUsageResponse)
+ return parse_response(electric_usage_response)
-def requestElectric():
+def request_electric():
return {
- "halfHour": doElectricRequest(RequestMode.halfHour),
- "hour": doElectricRequest(RequestMode.hour),
- "day": doElectricRequest(RequestMode.day),
- "month": doElectricRequest(RequestMode.month)
+ "halfHour": do_electric_request(RequestMode.halfHour),
+ "hour": do_electric_request(RequestMode.hour),
+ "day": do_electric_request(RequestMode.day),
+ "month": do_electric_request(RequestMode.month)
}
# Water
-def doWaterRequest(requestMode: RequestMode):
- waterUsageRequestJson['Mode'] = requestMode.value
+def do_water_request(requestMode: RequestMode):
+ water_usage_request_json['Mode'] = requestMode.value
if (requestMode in [RequestMode.hour]):
- waterUsageRequestJson['strDate'] = getYesterday()
- return performWaterRequest()
+ water_usage_request_json['strDate'] = get_yesterday_date()
+ return perform_water_request()
-def performWaterRequest():
- waterUsageResponse = requests.post(
- waterRequestEndpoint,
- cookies=usageRequestCookies,
- headers=genericRequestHeaders,
- json=waterUsageRequestJson
+def perform_water_request():
+ water_usage_response = requests.post(
+ water_request_endpoint,
+ cookies=usage_request_cookies,
+ headers=generic_request_headers,
+ json=water_usage_request_json
)
- return parseResponse(waterUsageResponse)
+ return parse_response(water_usage_response)
-def requestWater():
+def request_water():
return {
- "hour": doWaterRequest(RequestMode.hour),
- "day": doWaterRequest(RequestMode.day),
- "month": doWaterRequest(RequestMode.month)
+ "hour": do_water_request(RequestMode.hour),
+ "day": do_water_request(RequestMode.day),
+ "month": do_water_request(RequestMode.month)
}
# Utility methods
-def parseResponse(response):
- jsonResponse = json.loads(response.text.replace("\\\"", "\"").replace("\\\"", "\"").replace("\"{\"", "{\"").replace("}\"}", "}}"))['d']
- cleanResponse(jsonResponse)
+def parse_response(response):
+ json_response = json.loads(response.text.replace("\\\"", "\"").replace("\\\"", "\"").replace("\"{\"", "{\"").replace("}\"}", "}}"))['d']
+ clean_response(json_response)
return {
- "usageData": jsonResponse['objUsageGenerationResultSetTwo'], # Raw usage data for each timeframe
- "tentativeData": jsonResponse['getTentativeData'] # Accumulated usage data and predictions
+ "usageData": json_response['objUsageGenerationResultSetTwo'], # Raw usage data for each timeframe
+ "tentativeData": json_response['getTentativeData'] # Accumulated usage data and predictions
}
-def setupRequestParameters(parameters):
+def setup_request_params(parameters):
# Setup cookies and csrftoken to perform requests
- usageRequestCookies['ApplicationGatewayAffinityCORS'] = parameters['aga']
- usageRequestCookies['ApplicationGatewayAffinity'] = parameters['aga']
- usageRequestCookies['ASP.NET_SessionId'] = parameters['asi']
- usageRequestCookies['SCP'] = parameters['lt']
- genericRequestHeaders['csrftoken'] = parameters['ct']
+ usage_request_cookies['ApplicationGatewayAffinityCORS'] = parameters['aga']
+ usage_request_cookies['ApplicationGatewayAffinity'] = parameters['aga']
+ usage_request_cookies['ASP.NET_SessionId'] = parameters['asi']
+ usage_request_cookies['SCP'] = parameters['lt']
+ generic_request_headers['csrftoken'] = parameters['ct']
# Service calling method
-def requestUsageData(requestParameters):
- setupRequestParameters(requestParameters)
+def request_usage_data(request_params):
+ setup_request_params(request_params)
return {
- "electric": requestElectric(),
- "water": requestWater()
+ "electric": request_electric(),
+ "water": request_water()
}
-def getYesterday():
+def get_yesterday_date():
yesterday = date.today() - timedelta(days = 1)
return yesterday.strftime("%x")
-def cleanResponse(response: dict):
+def clean_response(response: dict):
# Remove the response keys that provide no information
- for key in uselessResponseKeys:
- removeKey(response, key)
+ for key in useless_response_keys:
+ remove_key(response, key)
-def removeKey(object, keyBeingRemoved):
+def remove_key(object, key_to_remove):
# If object is a dict, recursively search over data for keyBeingRemoved
if (isinstance(object, dict)):
- for listData in list(object):
- if listData == keyBeingRemoved:
- del object[keyBeingRemoved]
+ for list_data in list(object):
+ if list_data == key_to_remove:
+ del object[key_to_remove]
else:
- removeKey(object[listData], keyBeingRemoved)
+ remove_key(object[list_data], key_to_remove)
# If the object is a list, iterate over each object in the list
elif (isinstance(object, list)):
- for listData in list(object):
- removeKey(listData, keyBeingRemoved)
+ for list_data in list(object):
+ remove_key(list_data, key_to_remove)
diff --git a/app/request/login/login_request.py b/app/request/login/login_request.py
index 4bdd3f5..8b67cfb 100644
--- a/app/request/login/login_request.py
+++ b/app/request/login/login_request.py
@@ -1,14 +1,14 @@
import requests
import re
-from ...constants.constants import loginPageHeaders, loginRequestJson, genericRequestHeaders, loginPageUri, loginRequestEndpoint
+from ...constants.constants import login_page_headers, login_request_json, generic_request_headers, login_page_uri, login_request_endpoint
s = requests.session()
def login():
# Grab generated session keys from viewing the webpage
- aga, asi, ct = grabRequiredKeys()
+ aga, asi, ct = grab_required_keys()
# Perform a login request using keys found on the page and JSON data (credentials)
- lt = performLoginRequest(aga, asi, ct)
+ lt = perform_login_request(aga, asi, ct)
# Return the keys required to make endpoint calls
return {
"lt": lt,
@@ -18,40 +18,40 @@ def login():
}
-def grabRequiredKeys():
- loginPageResponse = s.get(loginPageUri, headers=loginPageHeaders)
- affinityMatcher = re.compile("CORS=(.+?);")
- affinityResults = affinityMatcher.search(loginPageResponse.headers['Set-Cookie'])
- appGatewayAffinity = affinityResults.group(1)
+def grab_required_keys():
+ login_page_response = s.get(login_page_uri, headers=login_page_headers)
+ affinity_matcher = re.compile("CORS=(.+?);")
+ affinity_results = affinity_matcher.search(login_page_response.headers['Set-Cookie'])
+ app_gateway_affinity = affinity_results.group(1)
- aspNetSessionIdMatcher = re.compile("ASP\.NET\_SessionId=(.+?);")
- aspnetResults = aspNetSessionIdMatcher.search(loginPageResponse.headers['Set-Cookie'])
- aspNetSessionId = aspnetResults.group(1)
+ asp_net_session_id_matcher = re.compile("ASP\.NET\_SessionId=(.+?);")
+ asp_net_results = asp_net_session_id_matcher.search(login_page_response.headers['Set-Cookie'])
+ asp_net_session_id = asp_net_results.group(1)
- csrfMatcher = re.compile("id=\"hdnCSRFToken\" value=\"(.+)\"")
- csrfResults = csrfMatcher.search(loginPageResponse.text)
- csrfToken = csrfResults.group(1)
+ csrf_matcher = re.compile("id=\"hdnCSRFToken\" value=\"(.+)\"")
+ csrf_results = csrf_matcher.search(login_page_response.text)
+ csrf_token = csrf_results.group(1)
s.cookies.clear()
- return appGatewayAffinity, aspNetSessionId, csrfToken
+ return app_gateway_affinity, asp_net_session_id, csrf_token
-def performLoginRequest(appGatewayAffinity, aspNetSessionId, csrfToken):
- loginRequestCookies = {
- 'ApplicationGatewayAffinityCORS': appGatewayAffinity,
- 'ApplicationGatewayAffinity': appGatewayAffinity,
- 'ASP.NET_SessionId': aspNetSessionId,
+def perform_login_request(app_gateway_affinity, asp_net_session_id, csrf_token):
+ login_request_cookies = {
+ 'ApplicationGatewayAffinityCORS': app_gateway_affinity,
+ 'ApplicationGatewayAffinity': app_gateway_affinity,
+ 'ASP.NET_SessionId': asp_net_session_id,
}
- genericRequestHeaders['csrftoken'] = csrfToken
+ generic_request_headers['csrftoken'] = csrf_token
- loginResponse = s.post(
- loginRequestEndpoint,
- cookies=loginRequestCookies,
- headers=genericRequestHeaders,
- json=loginRequestJson,
+ login_response = s.post(
+ login_request_endpoint,
+ cookies=login_request_cookies,
+ headers=generic_request_headers,
+ json=login_request_json,
)
- scpMatcher = re.compile("SCP=(.{36});")
- scpSearchResults = scpMatcher.search(loginResponse.headers['Set-Cookie'])
- loginToken = scpSearchResults.group(1)
+ scp_matcher = re.compile("SCP=(.{36});")
+ scp_results = scp_matcher.search(login_response.headers['Set-Cookie'])
+ login_token = scp_results.group(1)
s.cookies.clear()
- return loginToken
+ return login_token
generated by cgit v1.2.3 (git 2.25.1) at 2026-01-13 10:57:37 +0000