Skip to content

Commit

Permalink
feat: pulled out session and service_point
Browse files Browse the repository at this point in the history
  • Loading branch information
@Lash-L
Lash-L committed Feb 22, 2023
1 parent d28ea42 commit fa5863f
Show file tree
Hide file tree
Showing 5 changed files with 362 additions and 351 deletions.
1 change: 1 addition & 0 deletions .gitignore
View file
Original file line number Diff line number Diff line change
Expand Up @@ -138,3 +138,4 @@ dmypy.json

# Cython debug symbols
cython_debug/
src/southern_company_api/test.py
360 changes: 178 additions & 182 deletions src/southern_company_api/account.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,99 +57,101 @@ def __init__(
self.hourly_data: typing.Dict[str, HourlyEnergyUsage] = {}
self.daily_data: typing.Dict[str, DailyEnergyUsage] = {}
self.session = session
self.service_point_number = None

async def get_service_point_number(self, jwt: str) -> str:
async with self.session as session:
headers = {"Authorization": f"bearer {jwt}"}
# TODO: Is the /GPC for all customers or just GA power?
try:
async with session.get(
f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/"
f"getMPUBasicAccountInformation/{self.number}/GPC",
headers=headers,
) as resp:
headers = {
"Authorization": f"bearer {jwt}",
"content-type": "application/json, text/plain, */*",
}
# TODO: Is the /GPC for all customers or just GA power?
try:
async with self.session.get(
f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/"
f"getMPUBasicAccountInformation/{self.number}/GPC",
headers=headers,
) as resp:
try:
service_info = await resp.json()
except (ContentTypeError, json.JSONDecodeError) as err:
try:
service_info = await resp.json()
except (ContentTypeError, json.JSONDecodeError) as err:
try:
error_text = await resp.text()
except aiohttp.ClientError:
error_text = err.msg
raise CantReachSouthernCompany(
f"Incorrect mimetype while trying to get service point number. {error_text}"
) from err
error_text = await resp.text()
except aiohttp.ClientError:
error_text = err.msg
raise CantReachSouthernCompany(
f"Incorrect mimetype while trying to get service point number. error:{error_text} Response "
f"headers:{resp.headers} Your headers:{headers}"
) from err

# TODO: Test with multiple accounts
return service_info["Data"]["meterAndServicePoints"][0][
"servicePointNumber"
]
except aiohttp.ClientConnectorError as err:
raise CantReachSouthernCompany("Failed to connect to api") from err
# TODO: Test with multiple accounts
self.service_point_number = service_info["Data"][
"meterAndServicePoints"
][0]["servicePointNumber"]
return service_info["Data"]["meterAndServicePoints"][0][
"servicePointNumber"
]
except aiohttp.ClientConnectorError as err:
raise CantReachSouthernCompany("Failed to connect to api") from err

async def get_daily_data(
self, start_date: datetime.datetime, end_date: datetime.datetime, jwt: str
) -> List[DailyEnergyUsage]:
"""Available 24 hours after"""
"""This is not really tested yet."""
async with self.session as session:
headers = {"Authorization": f"bearer {jwt}"}
params = {
"accountNumber": self.number,
"startDate": start_date.strftime("%m/%d/%Y 12:00:00 AM"),
"endDate": end_date.strftime("%m/%d/%Y 11:59:59 PM"),
"OPCO": self.company.name,
"ServicePointNumber": await self.get_service_point_number(jwt),
"intervalBehavior": "Automatic",
}
async with session.get(
f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/"
f"MPUData/{self.number}/Daily",
headers=headers,
params=params,
) as resp:
if resp.status != 200:
raise UsageDataFailure(
f"Failed to get daily data: {resp.status} {headers}"
)
else:
headers = {"Authorization": f"bearer {jwt}"}
params = {
"accountNumber": self.number,
"startDate": start_date.strftime("%m/%d/%Y 12:00:00 AM"),
"endDate": end_date.strftime("%m/%d/%Y 11:59:59 PM"),
"OPCO": self.company.name,
"ServicePointNumber": self.service_point_number,
"intervalBehavior": "Automatic",
}
async with self.session.get(
f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/"
f"MPUData/{self.number}/Daily",
headers=headers,
params=params,
) as resp:
if resp.status != 200:
raise UsageDataFailure(
f"Failed to get daily data: {resp.status} {headers}"
)
else:
try:
response = await resp.json()
except (ContentTypeError, json.JSONDecodeError) as err:
try:
response = await resp.json()
except (ContentTypeError, json.JSONDecodeError) as err:
try:
error_text = await resp.text()
except aiohttp.ClientErrors:
error_text = err.msg
raise CantReachSouthernCompany(
f"Incorrect mimetype while trying to get daily data. {error_text}"
) from err
data = json.loads(response["Data"]["Data"])
day_maps = {}
dates = [date for date in data["xAxis"]["labels"]]
high_temps = [
temp["y"] for temp in data["series"]["highTemp"]["data"]
]
low_temps = [
temp["y"] for temp in data["series"]["lowTemp"]["data"]
]
for i, date in enumerate(dates):
day_maps[date] = DailyEnergyUsage(
# TODO: Determine timezone
date=datetime.datetime.strptime(date, "%Y-%m-%dT%H:%M:%S"),
usage=-1,
cost=1,
low_temp=low_temps[i],
high_temp=high_temps[i],
)
# TODO: Zip weekday and weekend to make it simpler.
for weekend_cost in data["series"]["weekdayCost"]["data"]:
day_maps[weekend_cost["name"]].cost = weekend_cost["y"]
for weekend_usage in data["series"]["weekdayUsage"]["data"]:
day_maps[weekend_usage["name"]].usage = weekend_usage["y"]
for weekday_cost in data["series"]["weekdayCost"]["data"]:
day_maps[weekday_cost["name"]].cost = weekday_cost["y"]
for weekday_usage in data["series"]["weekdayUsage"]["data"]:
day_maps[weekday_usage["name"]].usage = weekday_usage["y"]
return list(day_maps.values())
error_text = await resp.text()
except aiohttp.ClientError:
error_text = err.msg
raise CantReachSouthernCompany(
f"Incorrect mimetype while trying to get daily data. {error_text}"
) from err
data = json.loads(response["Data"]["Data"])
day_maps = {}
dates = [date for date in data["xAxis"]["labels"]]
high_temps = [temp["y"] for temp in data["series"]["highTemp"]["data"]]
low_temps = [temp["y"] for temp in data["series"]["lowTemp"]["data"]]
for i, date in enumerate(dates):
day_maps[date] = DailyEnergyUsage(
# TODO: Determine timezone
date=datetime.datetime.strptime(date, "%Y-%m-%dT%H:%M:%S"),
usage=-1,
cost=1,
low_temp=low_temps[i],
high_temp=high_temps[i],
)
# TODO: Zip weekday and weekend to make it simpler.
for weekend_cost in data["series"]["weekdayCost"]["data"]:
day_maps[weekend_cost["name"]].cost = weekend_cost["y"]
for weekend_usage in data["series"]["weekdayUsage"]["data"]:
day_maps[weekend_usage["name"]].usage = weekend_usage["y"]
for weekday_cost in data["series"]["weekdayCost"]["data"]:
day_maps[weekday_cost["name"]].cost = weekday_cost["y"]
for weekday_usage in data["series"]["weekdayUsage"]["data"]:
day_maps[weekday_usage["name"]].usage = weekday_usage["y"]
return list(day_maps.values())

async def get_hourly_data(
self, start_date: datetime.datetime, end_date: datetime.datetime, jwt: str
Expand All @@ -173,108 +175,102 @@ async def get_hourly_data(
continue
cur_date = cur_date + datetime.timedelta(days=35)
return return_data
async with self.session as session:
# Needs to check if the data already exist in self.hourly_data to avoid making an unneeded call.
headers = {"Authorization": f"bearer {jwt}"}
params = {
"accountNumber": self.number,
"startDate": start_date.strftime("%m/%d/%Y %H:%M:%S %p"),
"endDate": end_date.strftime("%m/%d/%Y %H:%M:%S %p"),
"OPCO": self.company.name,
"ServicePointNumber": await self.get_service_point_number(jwt),
"intervalBehavior": "Automatic",
}
async with session.get(
f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/"
f"MPUData/{self.number}/Hourly",
headers=headers,
params=params,
) as resp:
if resp.status != 200:
raise UsageDataFailure(
f"Failed to get hourly data: {resp.status} {headers}"
)
else:
# Needs to check if the data already exist in self.hourly_data to avoid making an unneeded call.
headers = {"Authorization": f"bearer {jwt}"}
params = {
"accountNumber": self.number,
"startDate": start_date.strftime("%m/%d/%Y %H:%M:%S %p"),
"endDate": end_date.strftime("%m/%d/%Y %H:%M:%S %p"),
"OPCO": self.company.name,
"ServicePointNumber": self.service_point_number,
"intervalBehavior": "Automatic",
}
async with self.session.get(
f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/"
f"MPUData/{self.number}/Hourly",
headers=headers,
params=params,
) as resp:
if resp.status != 200:
raise UsageDataFailure(
f"Failed to get hourly data: {resp.status} {headers}"
)
else:
try:
data = await resp.json()
except (ContentTypeError, json.JSONDecodeError) as err:
try:
data = await resp.json()
except (ContentTypeError, json.JSONDecodeError) as err:
try:
error_text = await resp.text()
except aiohttp.ClientError:
error_text = err.msg
raise CantReachSouthernCompany(
f"Incorrect mimetype while trying to get hourly data. {error_text}"
) from err
if data["Data"]["Data"] is None:
raise UsageDataFailure("Received no data back for usage.")
data = json.loads(data["Data"]["Data"])
return_dates = []
for date in data["xAxis"]["labels"]:
# TODO: Determine timezone
parsed_date = datetime.datetime.strptime(
date, "%Y-%m-%dT%H:%M:%S"
)
parsed_date = parsed_date.replace(
tzinfo=datetime.timezone(
datetime.timedelta(hours=-5), "EST"
)
)
self.hourly_data[date] = HourlyEnergyUsage(
time=parsed_date, usage=None, cost=None, temp=None
)
return_dates.append(self.hourly_data[date])
# costs and temps can be different lengths?
for cost in data["series"]["cost"]["data"]:
self.hourly_data[cost["name"]].cost = cost["y"]
for usage in data["series"]["usage"]["data"]:
self.hourly_data[usage["name"]].usage = usage["y"]
for temp in data["series"]["temp"]["data"]:
self.hourly_data[temp["name"]].temp = temp["y"]
return return_dates
error_text = await resp.text()
except aiohttp.ClientError:
error_text = err.msg
raise CantReachSouthernCompany(
f"Incorrect mimetype while trying to get hourly data. {error_text}"
) from err
if data["Data"]["Data"] is None:
raise UsageDataFailure("Received no data back for usage.")
data = json.loads(data["Data"]["Data"])
return_dates = []
for date in data["xAxis"]["labels"]:
# TODO: Determine timezone
parsed_date = datetime.datetime.strptime(date, "%Y-%m-%dT%H:%M:%S")
parsed_date = parsed_date.replace(
tzinfo=datetime.timezone(datetime.timedelta(hours=-5), "EST")
)
self.hourly_data[date] = HourlyEnergyUsage(
time=parsed_date, usage=None, cost=None, temp=None
)
return_dates.append(self.hourly_data[date])
# costs and temps can be different lengths?
for cost in data["series"]["cost"]["data"]:
self.hourly_data[cost["name"]].cost = cost["y"]
for usage in data["series"]["usage"]["data"]:
self.hourly_data[usage["name"]].usage = usage["y"]
for temp in data["series"]["temp"]["data"]:
self.hourly_data[temp["name"]].temp = temp["y"]
return return_dates

async def get_month_data(self, jwt: str) -> MonthlyUsage:
"""Gets monthly data such as usage so far"""
async with self.session as session:
headers = {"Authorization": f"bearer {jwt}"}
today = datetime.datetime.now()
first_of_month = today.replace(day=1)
params = {
"accountNumber": self.number,
"startDate": first_of_month.strftime("%m/%d/%Y 12:00:00 AM"),
"endDate": today.strftime("%m/%d/%Y 11:59:59 PM"),
"OPCO": self.company.name,
"ServicePointNumber": await self.get_service_point_number(jwt),
"intervalBehavior": "Automatic",
}
async with session.get(
f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/"
f"MPUData/{self.number}/Daily",
headers=headers,
params=params,
) as resp:
if resp.status != 200:
raise UsageDataFailure(
f"Failed to get month data: {resp.status} {headers}"
)
else:
headers = {"Authorization": f"bearer {jwt}"}
today = datetime.datetime.now()
first_of_month = today.replace(day=1)
params = {
"accountNumber": self.number,
"startDate": first_of_month.strftime("%m/%d/%Y 12:00:00 AM"),
"endDate": today.strftime("%m/%d/%Y 11:59:59 PM"),
"OPCO": self.company.name,
"ServicePointNumber": self.service_point_number,
"intervalBehavior": "Automatic",
}
async with self.session.get(
f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/"
f"MPUData/{self.number}/Daily",
headers=headers,
params=params,
) as resp:
if resp.status != 200:
raise UsageDataFailure(
f"Failed to get month data: {resp.status} {headers}"
)
else:
try:
connection = await resp.json()
except (ContentTypeError, json.JSONDecodeError) as err:
try:
connection = await resp.json()
except (ContentTypeError, json.JSONDecodeError) as err:
try:
error_text = await resp.text()
except aiohttp.ClientError:
error_text = err.msg
raise CantReachSouthernCompany(
f"Incorrect mimetype while trying to get month data. {error_text}"
) from err
data = connection["Data"]
return MonthlyUsage(
dollars_to_date=data["DollarsToDate"],
total_kwh_used=data["TotalkWhUsed"],
average_daily_usage=data["AverageDailyUsage"],
average_daily_cost=data["AverageDailyCost"],
projected_usage_low=data["ProjectedUsageLow"],
projected_usage_high=data["ProjectedUsageHigh"],
projected_bill_amount_low=data["ProjectedBillAmountLow"],
projected_bill_amount_high=data["ProjectedBillAmountHigh"],
)
error_text = await resp.text()
except aiohttp.ClientError:
error_text = err.msg
raise CantReachSouthernCompany(
f"Incorrect mimetype while trying to get month data. {error_text}"
) from err
data = connection["Data"]
return MonthlyUsage(
dollars_to_date=data["DollarsToDate"],
total_kwh_used=data["TotalkWhUsed"],
average_daily_usage=data["AverageDailyUsage"],
average_daily_cost=data["AverageDailyCost"],
projected_usage_low=data["ProjectedUsageLow"],
projected_usage_high=data["ProjectedUsageHigh"],
projected_bill_amount_low=data["ProjectedBillAmountLow"],
projected_bill_amount_high=data["ProjectedBillAmountHigh"],
)
Loading

0 comments on commit fa5863f

Please sign in to comment.