max/hass-fpl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

get back fpl original folder

Browse Source
This commit is contained in:
Yordan Suarez
2022-07-21 20:55:06 -04:00
parent cc610ab3e1
commit 25b5966dcb
22 changed files with 512 additions and 1458 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
custom_components/fpl/ProjectedBillSensor.py
View File

@@ -1,36 +0,0 @@
from .FplSensor import FplSensor
class FplProjectedBillSensor(FplSensor):
def __init__(self, hass, config, account):
FplSensor.__init__(self, hass, config, account, "Projected Bill")
@property
def state(self):
data = self.data
try:
if "budget_bill" in data.keys():
if data["budget_bill"]:
if "budget_billing_projected_bill" in data.keys():
self._state = data["budget_billing_projected_bill"]
else:
if "projected_bill" in data.keys():
self._state = data["projected_bill"]
except:
pass
return self._state
@property
def device_state_attributes(self):
"""Return the state attributes."""
if "budget_bill" in self.data.keys():
self.attr["budget_bill"] = self.data["budget_bill"]
return self._state
@property
def icon(self):
return "mdi:currency-usd"

0
custom_components/fpl1/TestSensor.py → custom_components/fpl/TestSensor.py
View File

0
custom_components/fpl1/__init__.py → custom_components/fpl/__init__.py
View File

0
custom_components/fpl1/config_flow.py → custom_components/fpl/config_flow.py
View File

2
custom_components/fpl1/const.py → custom_components/fpl/const.py
View File

@@ -3,7 +3,7 @@
NAME = "FPL Integration"
DOMAIN = "fpl"
DOMAIN_DATA = f"{DOMAIN}_data"
VERSION = "0.1.0"
VERSION = "0.0.1"
PLATFORMS = ["sensor"]
REQUIRED_FILES = [
".translations/en.json",

0
custom_components/fpl1/fplDataUpdateCoordinator.py → custom_components/fpl/fplDataUpdateCoordinator.py
View File

9
custom_components/fpl1/fplEntity.py → custom_components/fpl/fplEntity.py
View File

@@ -30,17 +30,16 @@ class FplEntity(CoordinatorEntity, SensorEntity):
DOMAIN, self.account, self.sensorName.lower().replace(" ", "")
)
@property
def name(self):
return f"{DOMAIN.upper()} {self.account} {self.sensorName}"
@property
def device_info(self):
return {
"identifiers": {(DOMAIN, self.account)},
"name": f"FPL Account {self.account}",
<<<<<<< HEAD:custom_components/fpl1/fplEntity.py
"model": VERSION,
=======
"model": "FPL Monitoring API",
"sw_version": VERSION,
>>>>>>> master:custom_components/fpl/fplEntity.py
"manufacturer": "Florida Power & Light",
}

304
custom_components/fpl/fplapi.py
View File

@@ -1,13 +1,11 @@
"""Custom FPl api client"""
import logging
from datetime import datetime
from datetime import datetime, timedelta
import sys
import json
import aiohttp
import async_timeout
import json
import sys
# from bs4 import BeautifulSoup
STATUS_CATEGORY_OPEN = "OPEN"
# Api login result
@@ -18,18 +16,40 @@ LOGIN_RESULT_UNAUTHORIZED = "UNAUTHORIZED"
LOGIN_RESULT_FAILURE = "FAILURE"
_LOGGER = logging.getLogger(__package__)
TIMEOUT = 30
TIMEOUT = 5
API_HOST = "https://www.fpl.com"
URL_LOGIN = API_HOST + "/api/resources/login"
URL_LOGOUT = API_HOST + "/api/resources/logout"
URL_RESOURCES_HEADER = API_HOST + "/api/resources/header"
URL_RESOURCES_ACCOUNT = API_HOST + "/api/resources/account/{account}"
URL_BUDGET_BILLING_GRAPH = (
API_HOST + "/api/resources/account/{account}/budgetBillingGraph"
)
URL_RESOURCES_PROJECTED_BILL = (
API_HOST
+ "/api/resources/account/{account}/projectedBill"
+ "?premiseNumber={premise}&lastBilledDate={lastBillDate}"
)
URL_ENERGY_SERVICE = (
API_HOST + "/dashboard-api/resources/account/{account}/energyService/{account}"
)
URL_APPLIANCE_USAGE = (
API_HOST + "/dashboard-api/resources/account/{account}/applianceUsage/{account}"
)
URL_BUDGET_BILLING_PREMISE_DETAILS = (
API_HOST + "/api/resources/account/{account}/budgetBillingGraph/premiseDetails"
)
URL_LOGIN = "https://www.fpl.com/api/resources/login"
URL_RESOURCES_HEADER = "https://www.fpl.com/api/resources/header"
URL_RESOURCES_ACCOUNT = "https://www.fpl.com/api/resources/account/{account}"
URL_RESOURCES_PROJECTED_BILL = "https://www.fpl.com/api/resources/account/{account}/projectedBill?premiseNumber={premise}&lastBilledDate={lastBillDate}"
ENROLLED = "ENROLLED"
NOTENROLLED = "NOTENROLLED"
class FplApi(object):
class FplApi:
"""A class for getting energy usage information from Florida Power & Light."""
def __init__(self, username, password, session):
@@ -39,7 +59,7 @@ class FplApi(object):
self._session = session
async def async_get_data(self) -> dict:
# self._session = aiohttp.ClientSession()
"""Get data from fpl api"""
data = {}
data["accounts"] = []
if await self.login() == LOGIN_RESULT_OK:
@@ -47,130 +67,151 @@ class FplApi(object):
data["accounts"] = accounts
for account in accounts:
accountData = await self.__async_get_data(account)
data[account] = accountData
account_data = await self.__async_get_data(account)
data[account] = account_data
await self.logout()
return data
async def login(self):
"""login into fpl"""
_LOGGER.info("Logging in")
# login and get account information
result = LOGIN_RESULT_OK
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(
URL_LOGIN, auth=aiohttp.BasicAuth(self._username, self._password)
)
js = json.loads(await response.text())
if response.status == 200:
_LOGGER.info("Logging Successful")
return LOGIN_RESULT_OK
if response.reason == "Unauthorized":
result = LOGIN_RESULT_UNAUTHORIZED
if response.status == 401:
_LOGGER.error("Logging Unauthorized")
json_data = json.loads(await response.text())
if js["messages"][0]["messageCode"] != "login.success":
_LOGGER.error(f"Logging Failure")
result = LOGIN_RESULT_FAILURE
if json_data["messageCode"] == LOGIN_RESULT_INVALIDUSER:
return LOGIN_RESULT_INVALIDUSER
_LOGGER.info(f"Logging Successful")
if json_data["messageCode"] == LOGIN_RESULT_INVALIDPASSWORD:
return LOGIN_RESULT_INVALIDPASSWORD
except Exception as e:
_LOGGER.error(f"Error {e} : {sys.exc_info()[0]}")
result = LOGIN_RESULT_FAILURE
except Exception as exception:
_LOGGER.error("Error %s : %s", exception, sys.exc_info()[0])
return LOGIN_RESULT_FAILURE
return result
return LOGIN_RESULT_FAILURE
async def logout(self):
"""Logging out from fpl"""
_LOGGER.info("Logging out")
URL = "https://www.fpl.com/api/resources/logout"
try:
async with async_timeout.timeout(TIMEOUT):
await self._session.get(URL)
await self._session.get(URL_LOGOUT)
except Exception:
pass
async def async_get_open_accounts(self):
_LOGGER.info(f"Getting accounts")
"""Getting open accounts"""
_LOGGER.info("Getting open accounts")
result = []
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(URL_RESOURCES_HEADER)
js = await response.json()
accounts = js["data"]["accounts"]["data"]["data"]
json_data = await response.json()
accounts = json_data["data"]["accounts"]["data"]["data"]
for account in accounts:
if account["statusCategory"] == STATUS_CATEGORY_OPEN:
result.append(account["accountNumber"])
except Exception as e:
_LOGGER.error(f"Getting accounts {e}")
# self._account_number = js["data"]["selectedAccount"]["data"]["accountNumber"]
# self._premise_number = js["data"]["selectedAccount"]["data"]["acctSecSettings"]["premiseNumber"]
except Exception:
_LOGGER.error("Getting accounts %s", sys.exc_info())
return result
async def __async_get_data(self, account) -> dict:
_LOGGER.info(f"Getting Data")
"""Get data from resources endpoint"""
_LOGGER.info("Getting Data")
data = {}
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(
URL_RESOURCES_ACCOUNT.format(account=account)
)
accountData = (await response.json())["data"]
account_data = (await response.json())["data"]
premise = accountData["premiseNumber"].zfill(9)
premise = account_data.get("premiseNumber").zfill(9)
data["meterSerialNo"] = account_data["meterSerialNo"]
# currentBillDate
currentBillDate = datetime.strptime(
accountData["currentBillDate"].replace("-", "").split("T")[0], "%Y%m%d"
account_data["currentBillDate"].replace("-", "").split("T")[0], "%Y%m%d"
).date()
# nextBillDate
nextBillDate = datetime.strptime(
accountData["nextBillDate"].replace("-", "").split("T")[0], "%Y%m%d"
account_data["nextBillDate"].replace("-", "").split("T")[0], "%Y%m%d"
).date()
data["current_bill_date"] = str(currentBillDate)
data["next_bill_date"] = str(nextBillDate)
today = datetime.now().date()
remaining = (nextBillDate - today).days
days = (today - currentBillDate).days
data["service_days"] = (nextBillDate - currentBillDate).days
data["as_of_days"] = days
data["remaining_days"] = remaining
data["as_of_days"] = (today - currentBillDate).days
data["remaining_days"] = (nextBillDate - today).days
# zip code
zip_code = accountData["serviceAddress"]["zip"]
# zip_code = accountData["serviceAddress"]["zip"]
# projected bill
pbData = await self.__getFromProjectedBill(account, premise, currentBillDate)
data.update(pbData)
# programs
programsData = accountData["programs"]["data"]
programsData = account_data["programs"]["data"]
programs = dict()
_LOGGER.info(f"Getting Programs")
_LOGGER.info("Getting Programs")
for program in programsData:
if "enrollmentStatus" in program.keys():
key = program["name"]
programs[key] = program["enrollmentStatus"] == ENROLLED
if programs["BBL"]:
# budget billing
data["budget_bill"] = True
bblData = await self.__getBBL_async(account, data)
data.update(bblData)
def hasProgram(programName) -> bool:
return programName in programs and programs[programName]
# Budget Billing program
if hasProgram("BBL"):
data["budget_bill"] = True
bbl_data = await self.__getBBL_async(account, data)
data.update(bbl_data)
else:
data["budget_bill"] = False
# Get data from energy service
data.update(
await self.__getDataFromEnergyService(account, premise, currentBillDate)
)
# Get data from energy service ( hourly )
# data.update(
# await self.__getDataFromEnergyServiceHourly(
# account, premise, currentBillDate
# )
# )
data.update(await self.__getDataFromApplianceUsage(account, currentBillDate))
return data
async def __getFromProjectedBill(self, account, premise, currentBillDate) -> dict:
"""get data from projected bill endpoint"""
data = {}
try:
@@ -196,24 +237,26 @@ class FplApi(object):
data["projected_bill"] = projectedBill
data["daily_avg"] = dailyAvg
data["avg_high_temp"] = avgHighTemp
except:
except Exception:
pass
return data
async def __getBBL_async(self, account, projectedBillData) -> dict:
_LOGGER.info(f"Getting budget billing data")
"""Get budget billing data"""
_LOGGER.info("Getting budget billing data")
data = {}
URL = "https://www.fpl.com/api/resources/account/{account}/budgetBillingGraph/premiseDetails"
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(URL.format(account=account))
response = await self._session.get(
URL_BUDGET_BILLING_PREMISE_DETAILS.format(account=account)
)
if response.status == 200:
r = (await response.json())["data"]
dataList = r["graphData"]
startIndex = len(dataList) - 1
# startIndex = len(dataList) - 1
billingCharge = 0
budgetBillDeferBalance = r["defAmt"]
@@ -235,28 +278,29 @@ class FplApi(object):
data["budget_billing_bill_to_date"] = bbAsOfDateAmt
data["budget_billing_projected_bill"] = float(projectedBudgetBill)
except:
pass
URL = "https://www.fpl.com/api/resources/account/{account}/budgetBillingGraph"
except Exception as e:
_LOGGER.error("Error getting BBL: %s", e)
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(URL.format(account=account))
response = await self._session.get(
URL_BUDGET_BILLING_GRAPH.format(account=account)
)
if response.status == 200:
r = (await response.json())["data"]
data["bill_to_date"] = float(r["eleAmt"])
data["defered_amount"] = float(r["defAmt"])
except:
pass
except Exception as e:
_LOGGER.error("Error getting BBL: %s", e)
return data
async def __getDataFromEnergyService(
self, account, premise, lastBilledDate
) -> dict:
_LOGGER.info(f"Getting data from energy service")
URL = "https://www.fpl.com/dashboard-api/resources/account/{account}/energyService/{account}"
_LOGGER.info("Getting data from energy service")
date = str(lastBilledDate.strftime("%m%d%Y"))
JSON = {
@@ -278,7 +322,9 @@ class FplApi(object):
data = {}
async with async_timeout.timeout(TIMEOUT):
response = await self._session.post(URL.format(account=account), json=JSON)
response = await self._session.post(
URL_ENERGY_SERVICE.format(account=account), json=JSON
)
if response.status == 200:
r = (await response.json())["data"]
dailyUsage = []
@@ -294,15 +340,21 @@ class FplApi(object):
):
dailyUsage.append(
{
"usage": daily.get("kwhUsed"),
"cost": daily.get("billingCharge"),
"date": daily.get("date"),
"max_temperature": daily.get(
"averageHighTemperature"
"usage": daily["kwhUsed"],
"cost": daily["billingCharge"],
# "date": daily["date"],
"max_temperature": daily["averageHighTemperature"],
"netDeliveredKwh": daily["netDeliveredKwh"]
if "netDeliveredKwh" in daily.keys()
else 0,
"netReceivedKwh": daily["netReceivedKwh"]
if "netReceivedKwh" in daily.keys()
else 0,
"readTime": datetime.fromisoformat(
daily[
"readTime"
] # 2022-02-25T00:00:00.000-05:00
),
"netDeliveredKwh": daily.get("netDeliveredKwh"),
"netReceivedKwh": daily.get("netReceivedKwh"),
"readTime": daily.get("readTime"),
}
)
# totalPowerUsage += int(daily["kwhUsed"])
@@ -310,23 +362,103 @@ class FplApi(object):
# data["total_power_usage"] = totalPowerUsage
data["daily_usage"] = dailyUsage
data["projectedKWH"] = r["CurrentUsage"].get("projectedKWH")
data["dailyAverageKWH"] = r["CurrentUsage"].get("dailyAverageKWH")
data["billToDateKWH"] = r["CurrentUsage"].get("billToDateKWH")
data["recMtrReading"] = r["CurrentUsage"].get("recMtrReading")
data["delMtrReading"] = r["CurrentUsage"].get("delMtrReading")
data["billStartDate"] = r["CurrentUsage"].get("billStartDate")
data["projectedKWH"] = r["CurrentUsage"]["projectedKWH"]
data["dailyAverageKWH"] = r["CurrentUsage"]["dailyAverageKWH"]
data["billToDateKWH"] = r["CurrentUsage"]["billToDateKWH"]
data["recMtrReading"] = r["CurrentUsage"]["recMtrReading"]
data["delMtrReading"] = r["CurrentUsage"]["delMtrReading"]
data["billStartDate"] = r["CurrentUsage"]["billStartDate"]
return data
async def __getDataFromEnergyServiceHourly(
self, account, premise, lastBilledDate
) -> dict:
_LOGGER.info("Getting data from energy service Hourly")
# date = str(lastBilledDate.strftime("%m%d%Y"))
date = str((datetime.now() - timedelta(days=1)).strftime("%m%d%Y"))
JSON = {
"status": 2,
"channel": "WEB",
"amrFlag": "Y",
"accountType": "RESIDENTIAL",
"revCode": "1",
"premiseNumber": premise,
"projectedBillFlag": False,
"billComparisionFlag": False,
"monthlyFlag": False,
"frequencyType": "Hourly",
"applicationPage": "resDashBoard",
"startDate": date,
}
data = {}
# now = homeassistant.util.dt.utcnow()
# now = datetime.now().astimezone()
# hour = now.hour
async with async_timeout.timeout(TIMEOUT):
response = await self._session.post(
URL_ENERGY_SERVICE.format(account=account), json=JSON
)
if response.status == 200:
r = (await response.json())["data"]
dailyUsage = []
# totalPowerUsage = 0
if "data" in r["HourlyUsage"]:
for daily in r["HourlyUsage"]["data"]:
if (
"kwhUsed" in daily.keys()
and "billingCharge" in daily.keys()
and "date" in daily.keys()
and "averageHighTemperature" in daily.keys()
):
dailyUsage.append(
{
"usage": daily["kwhUsed"],
"cost": daily["billingCharge"],
# "date": daily["date"],
"max_temperature": daily["averageHighTemperature"],
"netDeliveredKwh": daily["netDeliveredKwh"]
if "netDeliveredKwh" in daily.keys()
else 0,
"netReceivedKwh": daily["netReceivedKwh"]
if "netReceivedKwh" in daily.keys()
else 0,
"readTime": datetime.fromisoformat(
daily[
"readTime"
] # 2022-02-25T00:00:00.000-05:00
),
}
)
# totalPowerUsage += int(daily["kwhUsed"])
# data["total_power_usage"] = totalPowerUsage
data["daily_usage"] = dailyUsage
data["projectedKWH"] = r["HourlyUsage"]["projectedKWH"]
data["dailyAverageKWH"] = r["HourlyUsage"]["dailyAverageKWH"]
data["billToDateKWH"] = r["HourlyUsage"]["billToDateKWH"]
data["recMtrReading"] = r["HourlyUsage"]["recMtrReading"]
data["delMtrReading"] = r["HourlyUsage"]["delMtrReading"]
data["billStartDate"] = r["HourlyUsage"]["billStartDate"]
return data
async def __getDataFromApplianceUsage(self, account, lastBilledDate) -> dict:
_LOGGER.info(f"Getting data from applicance usage")
URL = "https://www.fpl.com/dashboard-api/resources/account/{account}/applianceUsage/{account}"
"""get data from appliance usage"""
_LOGGER.info("Getting data from appliance usage")
JSON = {"startDate": str(lastBilledDate.strftime("%m%d%Y"))}
data = {}
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.post(
URL.format(account=account), json=JSON
URL_APPLIANCE_USAGE.format(account=account), json=JSON
)
if response.status == 200:
electric = (await response.json())["data"]["electric"]
@@ -340,7 +472,7 @@ class FplApi(object):
rr = full
data[e["category"].replace(" ", "_")] = rr
except:
except Exception:
pass
return {"energy_percent_by_applicance": data}

3
custom_components/fpl1/manifest.json → custom_components/fpl/manifest.json
View File

@@ -12,6 +12,5 @@
"integrationhelper"
],
"version": "1.0.0",
"issue_tracker": "https://github.com/dotKrad/hass-fpl/issues",
"homeassistant": "2021.12.7"
"issue_tracker": "https://github.com/dotKrad/hass-fpl/issues"
}

12
custom_components/fpl1/sensor.py → custom_components/fpl/sensor.py
View File

@@ -23,6 +23,7 @@ from .sensor_ProjectedBillSensor import (
from .sensor_AverageDailySensor import (
DailyAverageSensor,
BudgetDailyAverageSensor,
ActualDailyAverageSensor,
)
from .sensor_DailyUsageSensor import (
FplDailyUsageKWHSensor,
@@ -30,10 +31,6 @@ from .sensor_DailyUsageSensor import (
FplDailyDeliveredKWHSensor,
FplDailyReceivedKWHSensor,
)
<<<<<<< HEAD:custom_components/fpl1/sensor.py
=======
>>>>>>> master:custom_components/fpl/sensor.py
from .const import DOMAIN
# from .TestSensor import TestSensor
@@ -59,10 +56,10 @@ async def async_setup_entry(hass, entry, async_add_devices):
# usage sensors
fpl_accounts.append(DailyAverageSensor(coordinator, entry, account))
fpl_accounts.append(BudgetDailyAverageSensor(coordinator, entry, account))
fpl_accounts.append(ActualDailyAverageSensor(coordinator, entry, account))
fpl_accounts.append(FplDailyUsageSensor(coordinator, entry, account))
fpl_accounts.append(FplDailyUsageKWHSensor(coordinator, entry, account))
fpl_accounts.append(FplDailyReceivedKWHSensor(coordinator, entry, account))
fpl_accounts.append(FplDailyDeliveredKWHSensor(coordinator, entry, account))
# date sensors
fpl_accounts.append(CurrentBillDateSensor(coordinator, entry, account))
@@ -75,9 +72,6 @@ async def async_setup_entry(hass, entry, async_add_devices):
fpl_accounts.append(ProjectedKWHSensor(coordinator, entry, account))
fpl_accounts.append(DailyAverageKWHSensor(coordinator, entry, account))
fpl_accounts.append(BillToDateKWHSensor(coordinator, entry, account))
fpl_accounts.append(NetReceivedKWHSensor(coordinator, entry, account))
fpl_accounts.append(NetDeliveredKWHSensor(coordinator, entry, account))
fpl_accounts.append(NetReceivedKWHSensor(coordinator, entry, account))
fpl_accounts.append(NetDeliveredKWHSensor(coordinator, entry, account))

33
custom_components/fpl/sensor_AllData.py
View File

@@ -1,33 +0,0 @@
from .fplEntity import FplEntity
class AllDataSensor(FplEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "")
@property
def state(self):
budget = self.getData("budget_bill")
budget_billing_projected_bill = self.getData("budget_billing_projected_bill")
if budget == True and budget_billing_projected_bill is not None:
return self.getData("budget_billing_projected_bill")
try:
self._state=self.getData("projected_bill")
except:
pass
return self._state
@property
def icon(self):
return "mdi:currency-usd"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Budget Projected Bill"
attributes["device_class"] = "monetary"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "$"
return attributes

52
custom_components/fpl1/sensor_AverageDailySensor.py → custom_components/fpl/sensor_AverageDailySensor.py
View File

@@ -17,36 +17,14 @@ class DailyAverageSensor(FplMoneyEntity):
if budget and budget_billing_projected_bill is not None:
return self.getData("budget_billing_daily_avg")
try:
self._state=self.getData("daily_avg")
except:
pass
return self._state
return self.getData("daily_avg")
<<<<<<< HEAD
=======
@property
def icon(self):
return "mdi:currency-usd"
<<<<<<< HEAD:custom_components/fpl1/sensor_AverageDailySensor.py
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
=======
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Daily Average"
attributes["device_class"] = "monetary"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "$"
return attributes
>>>>>>> master
>>>>>>> master:custom_components/fpl/sensor_AverageDailySensor.py
class BudgetDailyAverageSensor(FplMoneyEntity):
"""budget daily average sensor"""
@@ -56,22 +34,14 @@ class BudgetDailyAverageSensor(FplMoneyEntity):
@property
def state(self):
try:
self._state= self.getData("budget_billing_daily_avg")
except:
pass
return self._state
return self.getData("budget_billing_daily_avg")
<<<<<<< HEAD:custom_components/fpl1/sensor_AverageDailySensor.py
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
=======
<<<<<<< HEAD
>>>>>>> master:custom_components/fpl/sensor_AverageDailySensor.py
class ActualDailyAverageSensor(FplMoneyEntity):
"""Actual daily average sensor"""
@@ -82,27 +52,9 @@ class ActualDailyAverageSensor(FplMoneyEntity):
@property
def state(self):
return self.getData("daily_avg")
<<<<<<< HEAD:custom_components/fpl1/sensor_AverageDailySensor.py
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
=======
=======
@property
def icon(self):
return "mdi:currency-usd"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Budget Daily Average"
attributes["device_class"] = "monitary"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "$"
return attributes
>>>>>>> master
>>>>>>> master:custom_components/fpl/sensor_AverageDailySensor.py

156
custom_components/fpl/sensor_DailyUsageSensor.py
View File

@@ -1,7 +1,12 @@
"""Daily Usage Sensors"""
from homeassistant.components.sensor import STATE_CLASS_TOTAL_INCREASING
from datetime import timedelta
from .fplEntity import FplEnergyEntity, FplMoneyEntity
class FplDailyUsageSensor(FplMoneyEntity):
"""Daily Usage Cost Sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Usage")
@@ -9,37 +14,25 @@ class FplDailyUsageSensor(FplMoneyEntity):
def state(self):
data = self.getData("daily_usage")
try:
self._state = data[-1]["cost"]
except:
pass
return self._state
if data is not None and len(data) > 0 and "cost" in data[-1].keys():
return data[-1]["cost"]
def defineAttributes(self):
return None
def customAttributes(self):
"""Return the state attributes."""
data = self.getData("daily_usage")
attributes = {}
attributes["friendly_name"] = "Daily Usage"
attributes["device_class"] = "monetary"
attributes["state_class"] = "total_increasing"
attributes["unit_of_measurement"] = "$"
if data is not None:
if (
(len(data) > 0)
and (data[-1] is not None)
and (data[-1]["readTime"] is not None)
):
# attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
if data is not None and len(data) > 0 and "readTime" in data[-1].keys():
attributes["date"] = data[-1]["readTime"]
if (
(len(data) > 1)
and (data[-2] is not None)
and (data[-2]["readTime"] is not None)
):
attributes["last_reset"] = data[-2]["readTime"]
return attributes
class FplDailyUsageKWHSensor(FplEnergyEntity):
"""Daily Usage Kwh Sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Usage KWH")
@@ -47,126 +40,71 @@ class FplDailyUsageKWHSensor(FplEnergyEntity):
def state(self):
data = self.getData("daily_usage")
try:
self._state = data[-1]["usage"]
except:
pass
return self._state
if data is not None and len(data) > 0 and "usage" in data[-1].keys():
return data[-1]["usage"]
def defineAttributes(self):
return None
def customAttributes(self):
"""Return the state attributes."""
data = self.getData("daily_usage")
date = data[-1]["readTime"]
last_reset = date - timedelta(days=1)
attributes = {}
attributes["friendly_name"] = "Daily Usage"
attributes["device_class"] = "energy"
attributes["state_class"] = "total_increasing"
attributes["unit_of_measurement"] = "kWh"
if data is not None:
if (
(len(data) > 0)
and (data[-1] is not None)
and (data[-1]["readTime"] is not None)
):
attributes["date"] = data[-1]["readTime"]
if (
(len(data) > 1)
and (data[-2] is not None)
and (data[-2]["readTime"] is not None)
):
attributes["last_reset"] = data[-2]["readTime"]
# attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
attributes["date"] = date
# attributes["last_reset"] = last_reset
return attributes
@property
def icon(self):
return "mdi:flash"
class FplDailyReceivedKWHSensor(FplEnergyEntity):
"""daily received Kwh sensor"""
class FplDailyReceivedKWHSensor(FplEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Received KWH")
@property
def state(self):
data = self.getData("daily_usage")
try:
self._state = data[-1]["netReceivedKwh"]
except:
pass
return self._state
if data is not None and len(data) > 0 and "netReceivedKwh" in data[-1].keys():
return data[-1]["netReceivedKwh"]
return 0
def defineAttributes(self):
def customAttributes(self):
"""Return the state attributes."""
data = self.getData("daily_usage")
date = data[-1]["readTime"]
last_reset = date - timedelta(days=1)
attributes = {}
attributes["friendly_name"] = "Daily Return to Grid"
attributes["device_class"] = "energy"
attributes["state_class"] = "total_increasing"
attributes["unit_of_measurement"] = "kWh"
if data is not None:
if (
(len(data) > 0)
and (data[-1] is not None)
and (data[-1]["readTime"] is not None)
):
attributes["date"] = data[-1]["readTime"]
if (
(len(data) > 1)
and (data[-2] is not None)
and (data[-2]["readTime"] is not None)
):
attributes["last_reset"] = data[-2]["readTime"]
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
attributes["date"] = date
attributes["last_reset"] = last_reset
return attributes
@property
def icon(self):
return "mdi:flash"
class FplDailyDeliveredKWHSensor(FplEnergyEntity):
"""daily delivered Kwh sensor"""
class FplDailyDeliveredKWHSensor(FplEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Delivered KWH")
@property
def state(self):
data = self.getData("daily_usage")
try:
self._state = data[-1]["netDeliveredKwh"]
except:
pass
return self._state
if data is not None and len(data) > 0 and "netDeliveredKwh" in data[-1].keys():
return data[-1]["netDeliveredKwh"]
return 0
def defineAttributes(self):
def customAttributes(self):
"""Return the state attributes."""
data = self.getData("daily_usage")
date = data[-1]["readTime"]
last_reset = date - timedelta(days=1)
<<<<<<< HEAD
return {}
=======
attributes = {}
attributes["friendly_name"] = "Daily Consumption"
attributes["device_class"] = "energy"
attributes["state_class"] = "total_increasing"
attributes["unit_of_measurement"] = "kWh"
if data is not None:
if (
(len(data) > 0)
and (data[-1] is not None)
and (data[-1]["readTime"] is not None)
):
attributes["date"] = data[-1]["readTime"]
if (
(len(data) > 1)
and (data[-2] is not None)
and (data[-2]["readTime"] is not None)
):
attributes["last_reset"] = data[-2]["readTime"]
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
attributes["date"] = date
attributes["last_reset"] = last_reset
return attributes
@property
def icon(self):
return "mdi:flash"
>>>>>>> master

48
custom_components/fpl/sensor_DatesSensor.py Normal file
View File

@@ -0,0 +1,48 @@
"""dates sensors"""
import datetime
from .fplEntity import FplDateEntity, FplDayEntity
class CurrentBillDateSensor(FplDateEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Current Bill Date")
@property
def state(self):
return datetime.date.fromisoformat(self.getData("current_bill_date"))
class NextBillDateSensor(FplDateEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Next Bill Date")
@property
def state(self):
return datetime.date.fromisoformat(self.getData("next_bill_date"))
class ServiceDaysSensor(FplDayEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Service Days")
@property
def state(self):
return self.getData("service_days")
class AsOfDaysSensor(FplDayEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "As Of Days")
@property
def state(self):
return self.getData("as_of_days")
class RemainingDaysSensor(FplDayEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Remaining Days")
@property
def state(self):
return self.getData("remaining_days")

90
custom_components/fpl/sensor_KWHSensor.py Normal file
View File

@@ -0,0 +1,90 @@
"""energy sensors"""
from datetime import date, timedelta
import datetime
from homeassistant.components.sensor import (
STATE_CLASS_TOTAL_INCREASING,
STATE_CLASS_TOTAL,
)
from .fplEntity import FplEnergyEntity
class ProjectedKWHSensor(FplEnergyEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Projected KWH")
@property
def state(self):
return self.getData("projectedKWH")
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
class DailyAverageKWHSensor(FplEnergyEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Average KWH")
@property
def state(self):
return self.getData("dailyAverageKWH")
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
class BillToDateKWHSensor(FplEnergyEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Bill To Date KWH")
@property
def state(self):
return self.getData("billToDateKWH")
def customAttributes(self):
"""Return the state attributes."""
# data = self.getData("daily_usage")
# date = data[-1]["readTime"]
asOfDays = self.getData("as_of_days")
last_reset = date.today() - timedelta(days=asOfDays)
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
attributes["last_reset"] = last_reset
return attributes
class NetReceivedKWHSensor(FplEnergyEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Received Meter Reading KWH")
@property
def state(self):
return self.getData("recMtrReading")
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
return attributes
class NetDeliveredKWHSensor(FplEnergyEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Delivered Meter Reading KWH")
@property
def state(self):
return self.getData("delMtrReading")
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
return attributes

96
custom_components/fpl/sensor_ProjectedBillSensor.py Normal file
View File

@@ -0,0 +1,96 @@
"""Projected bill sensors"""
from homeassistant.components.sensor import (
# STATE_CLASS_TOTAL_INCREASING,
STATE_CLASS_TOTAL,
)
from .fplEntity import FplMoneyEntity
class FplProjectedBillSensor(FplMoneyEntity):
"""projected bill sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Projected Bill")
@property
def native_value(self):
budget = self.getData("budget_bill")
budget_billing_projected_bill = self.getData("budget_billing_projected_bill")
if budget and budget_billing_projected_bill is not None:
return self.getData("budget_billing_projected_bill")
return self.getData("projected_bill")
"""
@property
def state(self):
budget = self.getData("budget_bill")
budget_billing_projected_bill = self.getData("budget_billing_projected_bill")
if budget and budget_billing_projected_bill is not None:
return self.getData("budget_billing_projected_bill")
return self.getData("projected_bill")
"""
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
attributes["budget_bill"] = self.getData("budget_bill")
return attributes
# Defered Amount
class DeferedAmountSensor(FplMoneyEntity):
"""Defered amount sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Defered Amount")
@property
def state(self):
if self.getData("budget_bill"):
return self.getData("defered_amount")
return 0
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
class ProjectedBudgetBillSensor(FplMoneyEntity):
"""projected budget bill sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Projected Budget Bill")
@property
def state(self):
return self.getData("budget_billing_projected_bill")
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
class ProjectedActualBillSensor(FplMoneyEntity):
"""projeted actual bill sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Projected Actual Bill")
@property
def state(self):
return self.getData("projected_bill")
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes

0
custom_components/fpl1/translations/en.json → custom_components/fpl/translations/en.json
View File

478
custom_components/fpl1/fplapi.py
View File

@@ -1,478 +0,0 @@
"""Custom FPl api client"""
import logging
from datetime import datetime, timedelta
import sys
import json
import aiohttp
import async_timeout
STATUS_CATEGORY_OPEN = "OPEN"
# Api login result
LOGIN_RESULT_OK = "OK"
LOGIN_RESULT_INVALIDUSER = "NOTVALIDUSER"
LOGIN_RESULT_INVALIDPASSWORD = "FAILEDPASSWORD"
LOGIN_RESULT_UNAUTHORIZED = "UNAUTHORIZED"
LOGIN_RESULT_FAILURE = "FAILURE"
_LOGGER = logging.getLogger(__package__)
TIMEOUT = 5
API_HOST = "https://www.fpl.com"
URL_LOGIN = API_HOST + "/api/resources/login"
URL_LOGOUT = API_HOST + "/api/resources/logout"
URL_RESOURCES_HEADER = API_HOST + "/api/resources/header"
URL_RESOURCES_ACCOUNT = API_HOST + "/api/resources/account/{account}"
URL_BUDGET_BILLING_GRAPH = (
API_HOST + "/api/resources/account/{account}/budgetBillingGraph"
)
URL_RESOURCES_PROJECTED_BILL = (
API_HOST
+ "/api/resources/account/{account}/projectedBill"
+ "?premiseNumber={premise}&lastBilledDate={lastBillDate}"
)
URL_ENERGY_SERVICE = (
API_HOST + "/dashboard-api/resources/account/{account}/energyService/{account}"
)
URL_APPLIANCE_USAGE = (
API_HOST + "/dashboard-api/resources/account/{account}/applianceUsage/{account}"
)
URL_BUDGET_BILLING_PREMISE_DETAILS = (
API_HOST + "/api/resources/account/{account}/budgetBillingGraph/premiseDetails"
)
ENROLLED = "ENROLLED"
NOTENROLLED = "NOTENROLLED"
class FplApi:
"""A class for getting energy usage information from Florida Power & Light."""
def __init__(self, username, password, session):
"""Initialize the data retrieval. Session should have BasicAuth flag set."""
self._username = username
self._password = password
self._session = session
async def async_get_data(self) -> dict:
"""Get data from fpl api"""
data = {}
data["accounts"] = []
if await self.login() == LOGIN_RESULT_OK:
accounts = await self.async_get_open_accounts()
data["accounts"] = accounts
for account in accounts:
account_data = await self.__async_get_data(account)
data[account] = account_data
await self.logout()
return data
async def login(self):
"""login into fpl"""
_LOGGER.info("Logging in")
# login and get account information
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(
URL_LOGIN, auth=aiohttp.BasicAuth(self._username, self._password)
)
if response.status == 200:
_LOGGER.info("Logging Successful")
return LOGIN_RESULT_OK
if response.status == 401:
_LOGGER.error("Logging Unauthorized")
json_data = json.loads(await response.text())
if json_data["messageCode"] == LOGIN_RESULT_INVALIDUSER:
return LOGIN_RESULT_INVALIDUSER
if json_data["messageCode"] == LOGIN_RESULT_INVALIDPASSWORD:
return LOGIN_RESULT_INVALIDPASSWORD
except Exception as exception:
_LOGGER.error("Error %s : %s", exception, sys.exc_info()[0])
return LOGIN_RESULT_FAILURE
return LOGIN_RESULT_FAILURE
async def logout(self):
"""Logging out from fpl"""
_LOGGER.info("Logging out")
try:
async with async_timeout.timeout(TIMEOUT):
await self._session.get(URL_LOGOUT)
except Exception:
pass
async def async_get_open_accounts(self):
"""Getting open accounts"""
_LOGGER.info("Getting open accounts")
result = []
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(URL_RESOURCES_HEADER)
json_data = await response.json()
accounts = json_data["data"]["accounts"]["data"]["data"]
for account in accounts:
if account["statusCategory"] == STATUS_CATEGORY_OPEN:
result.append(account["accountNumber"])
except Exception:
_LOGGER.error("Getting accounts %s", sys.exc_info())
return result
async def __async_get_data(self, account) -> dict:
"""Get data from resources endpoint"""
_LOGGER.info("Getting Data")
data = {}
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(
URL_RESOURCES_ACCOUNT.format(account=account)
)
account_data = (await response.json())["data"]
premise = account_data.get("premiseNumber").zfill(9)
data["meterSerialNo"] = account_data["meterSerialNo"]
# currentBillDate
currentBillDate = datetime.strptime(
account_data["currentBillDate"].replace("-", "").split("T")[0], "%Y%m%d"
).date()
# nextBillDate
nextBillDate = datetime.strptime(
account_data["nextBillDate"].replace("-", "").split("T")[0], "%Y%m%d"
).date()
data["current_bill_date"] = str(currentBillDate)
data["next_bill_date"] = str(nextBillDate)
today = datetime.now().date()
data["service_days"] = (nextBillDate - currentBillDate).days
data["as_of_days"] = (today - currentBillDate).days
data["remaining_days"] = (nextBillDate - today).days
# zip code
# zip_code = accountData["serviceAddress"]["zip"]
# projected bill
pbData = await self.__getFromProjectedBill(account, premise, currentBillDate)
data.update(pbData)
# programs
programsData = account_data["programs"]["data"]
programs = dict()
_LOGGER.info("Getting Programs")
for program in programsData:
if "enrollmentStatus" in program.keys():
key = program["name"]
programs[key] = program["enrollmentStatus"] == ENROLLED
def hasProgram(programName) -> bool:
return programName in programs and programs[programName]
# Budget Billing program
if hasProgram("BBL"):
data["budget_bill"] = True
bbl_data = await self.__getBBL_async(account, data)
data.update(bbl_data)
else:
data["budget_bill"] = False
# Get data from energy service
data.update(
await self.__getDataFromEnergyService(account, premise, currentBillDate)
)
# Get data from energy service ( hourly )
# data.update(
# await self.__getDataFromEnergyServiceHourly(
# account, premise, currentBillDate
# )
# )
data.update(await self.__getDataFromApplianceUsage(account, currentBillDate))
return data
async def __getFromProjectedBill(self, account, premise, currentBillDate) -> dict:
"""get data from projected bill endpoint"""
data = {}
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(
URL_RESOURCES_PROJECTED_BILL.format(
account=account,
premise=premise,
lastBillDate=currentBillDate.strftime("%m%d%Y"),
)
)
if response.status == 200:
projectedBillData = (await response.json())["data"]
billToDate = float(projectedBillData["billToDate"])
projectedBill = float(projectedBillData["projectedBill"])
dailyAvg = float(projectedBillData["dailyAvg"])
avgHighTemp = int(projectedBillData["avgHighTemp"])
data["bill_to_date"] = billToDate
data["projected_bill"] = projectedBill
data["daily_avg"] = dailyAvg
data["avg_high_temp"] = avgHighTemp
except Exception:
pass
return data
async def __getBBL_async(self, account, projectedBillData) -> dict:
"""Get budget billing data"""
_LOGGER.info("Getting budget billing data")
data = {}
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(
URL_BUDGET_BILLING_PREMISE_DETAILS.format(account=account)
)
if response.status == 200:
r = (await response.json())["data"]
dataList = r["graphData"]
# startIndex = len(dataList) - 1
billingCharge = 0
budgetBillDeferBalance = r["defAmt"]
projectedBill = projectedBillData["projected_bill"]
asOfDays = projectedBillData["as_of_days"]
for det in dataList:
billingCharge += det["actuallBillAmt"]
calc1 = (projectedBill + billingCharge) / 12
calc2 = (1 / 12) * (budgetBillDeferBalance)
projectedBudgetBill = round(calc1 + calc2, 2)
bbDailyAvg = round(projectedBudgetBill / 30, 2)
bbAsOfDateAmt = round(projectedBudgetBill / 30 * asOfDays, 2)
data["budget_billing_daily_avg"] = bbDailyAvg
data["budget_billing_bill_to_date"] = bbAsOfDateAmt
data["budget_billing_projected_bill"] = float(projectedBudgetBill)
except Exception as e:
_LOGGER.error("Error getting BBL: %s", e)
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.get(
URL_BUDGET_BILLING_GRAPH.format(account=account)
)
if response.status == 200:
r = (await response.json())["data"]
data["bill_to_date"] = float(r["eleAmt"])
data["defered_amount"] = float(r["defAmt"])
except Exception as e:
_LOGGER.error("Error getting BBL: %s", e)
return data
async def __getDataFromEnergyService(
self, account, premise, lastBilledDate
) -> dict:
_LOGGER.info("Getting data from energy service")
date = str(lastBilledDate.strftime("%m%d%Y"))
JSON = {
"recordCount": 24,
"status": 2,
"channel": "WEB",
"amrFlag": "Y",
"accountType": "RESIDENTIAL",
"revCode": "1",
"premiseNumber": premise,
"projectedBillFlag": True,
"billComparisionFlag": True,
"monthlyFlag": True,
"frequencyType": "Daily",
"lastBilledDate": date,
"applicationPage": "resDashBoard",
}
data = {}
async with async_timeout.timeout(TIMEOUT):
response = await self._session.post(
URL_ENERGY_SERVICE.format(account=account), json=JSON
)
if response.status == 200:
r = (await response.json())["data"]
dailyUsage = []
# totalPowerUsage = 0
if "data" in r["DailyUsage"]:
for daily in r["DailyUsage"]["data"]:
if (
"kwhUsed" in daily.keys()
and "billingCharge" in daily.keys()
and "date" in daily.keys()
and "averageHighTemperature" in daily.keys()
):
dailyUsage.append(
{
"usage": daily["kwhUsed"],
"cost": daily["billingCharge"],
# "date": daily["date"],
"max_temperature": daily["averageHighTemperature"],
"netDeliveredKwh": daily["netDeliveredKwh"]
if "netDeliveredKwh" in daily.keys()
else 0,
"netReceivedKwh": daily["netReceivedKwh"]
if "netReceivedKwh" in daily.keys()
else 0,
"readTime": datetime.fromisoformat(
daily[
"readTime"
] # 2022-02-25T00:00:00.000-05:00
),
}
)
# totalPowerUsage += int(daily["kwhUsed"])
# data["total_power_usage"] = totalPowerUsage
data["daily_usage"] = dailyUsage
data["projectedKWH"] = r["CurrentUsage"]["projectedKWH"]
data["dailyAverageKWH"] = r["CurrentUsage"]["dailyAverageKWH"]
data["billToDateKWH"] = r["CurrentUsage"]["billToDateKWH"]
data["recMtrReading"] = r["CurrentUsage"]["recMtrReading"]
data["delMtrReading"] = r["CurrentUsage"]["delMtrReading"]
data["billStartDate"] = r["CurrentUsage"]["billStartDate"]
return data
async def __getDataFromEnergyServiceHourly(
self, account, premise, lastBilledDate
) -> dict:
_LOGGER.info("Getting data from energy service Hourly")
# date = str(lastBilledDate.strftime("%m%d%Y"))
date = str((datetime.now() - timedelta(days=1)).strftime("%m%d%Y"))
JSON = {
"status": 2,
"channel": "WEB",
"amrFlag": "Y",
"accountType": "RESIDENTIAL",
"revCode": "1",
"premiseNumber": premise,
"projectedBillFlag": False,
"billComparisionFlag": False,
"monthlyFlag": False,
"frequencyType": "Hourly",
"applicationPage": "resDashBoard",
"startDate": date,
}
data = {}
# now = homeassistant.util.dt.utcnow()
# now = datetime.now().astimezone()
# hour = now.hour
async with async_timeout.timeout(TIMEOUT):
response = await self._session.post(
URL_ENERGY_SERVICE.format(account=account), json=JSON
)
if response.status == 200:
r = (await response.json())["data"]
dailyUsage = []
# totalPowerUsage = 0
if "data" in r["HourlyUsage"]:
for daily in r["HourlyUsage"]["data"]:
if (
"kwhUsed" in daily.keys()
and "billingCharge" in daily.keys()
and "date" in daily.keys()
and "averageHighTemperature" in daily.keys()
):
dailyUsage.append(
{
"usage": daily["kwhUsed"],
"cost": daily["billingCharge"],
# "date": daily["date"],
"max_temperature": daily["averageHighTemperature"],
"netDeliveredKwh": daily["netDeliveredKwh"]
if "netDeliveredKwh" in daily.keys()
else 0,
"netReceivedKwh": daily["netReceivedKwh"]
if "netReceivedKwh" in daily.keys()
else 0,
"readTime": datetime.fromisoformat(
daily[
"readTime"
] # 2022-02-25T00:00:00.000-05:00
),
}
)
# totalPowerUsage += int(daily["kwhUsed"])
# data["total_power_usage"] = totalPowerUsage
data["daily_usage"] = dailyUsage
data["projectedKWH"] = r["HourlyUsage"]["projectedKWH"]
data["dailyAverageKWH"] = r["HourlyUsage"]["dailyAverageKWH"]
data["billToDateKWH"] = r["HourlyUsage"]["billToDateKWH"]
data["recMtrReading"] = r["HourlyUsage"]["recMtrReading"]
data["delMtrReading"] = r["HourlyUsage"]["delMtrReading"]
data["billStartDate"] = r["HourlyUsage"]["billStartDate"]
return data
async def __getDataFromApplianceUsage(self, account, lastBilledDate) -> dict:
"""get data from appliance usage"""
_LOGGER.info("Getting data from appliance usage")
JSON = {"startDate": str(lastBilledDate.strftime("%m%d%Y"))}
data = {}
try:
async with async_timeout.timeout(TIMEOUT):
response = await self._session.post(
URL_APPLIANCE_USAGE.format(account=account), json=JSON
)
if response.status == 200:
electric = (await response.json())["data"]["electric"]
full = 100
for e in electric:
rr = round(float(e["percentageDollar"]))
if rr < full:
full = full - rr
else:
rr = full
data[e["category"].replace(" ", "_")] = rr
except Exception:
pass
return {"energy_percent_by_applicance": data}

110
custom_components/fpl1/sensor_DailyUsageSensor.py
View File

@@ -1,110 +0,0 @@
"""Daily Usage Sensors"""
from homeassistant.components.sensor import STATE_CLASS_TOTAL_INCREASING
from datetime import timedelta
from .fplEntity import FplEnergyEntity, FplMoneyEntity
class FplDailyUsageSensor(FplMoneyEntity):
"""Daily Usage Cost Sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Usage")
@property
def state(self):
data = self.getData("daily_usage")
if data is not None and len(data) > 0 and "cost" in data[-1].keys():
return data[-1]["cost"]
return None
def customAttributes(self):
"""Return the state attributes."""
data = self.getData("daily_usage")
attributes = {}
# attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
if data is not None and len(data) > 0 and "readTime" in data[-1].keys():
attributes["date"] = data[-1]["readTime"]
return attributes
class FplDailyUsageKWHSensor(FplEnergyEntity):
"""Daily Usage Kwh Sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Usage KWH")
@property
def state(self):
data = self.getData("daily_usage")
if data is not None and len(data) > 0 and "usage" in data[-1].keys():
return data[-1]["usage"]
return None
def customAttributes(self):
"""Return the state attributes."""
data = self.getData("daily_usage")
date = data[-1]["readTime"]
last_reset = date - timedelta(days=1)
attributes = {}
# attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
attributes["date"] = date
# attributes["last_reset"] = last_reset
return attributes
class FplDailyReceivedKWHSensor(FplEnergyEntity):
"""daily received Kwh sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Received KWH")
@property
def state(self):
data = self.getData("daily_usage")
if data is not None and len(data) > 0 and "netReceivedKwh" in data[-1].keys():
return data[-1]["netReceivedKwh"]
return 0
def customAttributes(self):
"""Return the state attributes."""
data = self.getData("daily_usage")
date = data[-1]["readTime"]
last_reset = date - timedelta(days=1)
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
attributes["date"] = date
attributes["last_reset"] = last_reset
return attributes
class FplDailyDeliveredKWHSensor(FplEnergyEntity):
"""daily delivered Kwh sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Delivered KWH")
@property
def state(self):
data = self.getData("daily_usage")
if data is not None and len(data) > 0 and "netDeliveredKwh" in data[-1].keys():
return data[-1]["netDeliveredKwh"]
return 0
def customAttributes(self):
"""Return the state attributes."""
data = self.getData("daily_usage")
date = data[-1]["readTime"]
last_reset = date - timedelta(days=1)
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
attributes["date"] = date
attributes["last_reset"] = last_reset
return attributes

153
custom_components/fpl1/sensor_DatesSensor.py
View File

@@ -1,153 +0,0 @@
<<<<<<< HEAD:custom_components/fpl1/sensor_DatesSensor.py
"""dates sensors"""
import datetime
from .fplEntity import FplDateEntity, FplDayEntity
=======
<<<<<<< HEAD
from .fplEntity import FplDateEntity
>>>>>>> master:custom_components/fpl/sensor_DatesSensor.py
=======
from .fplEntity import FplEntity
import datetime
>>>>>>> master
class CurrentBillDateSensor(FplDateEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Billing Current Date")
@property
def state(self):
<<<<<<< HEAD:custom_components/fpl1/sensor_DatesSensor.py
return datetime.date.fromisoformat(self.getData("current_bill_date"))
=======
try:
self._state= datetime.date.fromisoformat(self.getData("current_bill_date"))
except:
pass
return self._state
>>>>>>> master:custom_components/fpl/sensor_DatesSensor.py
<<<<<<< HEAD
=======
@property
def icon(self):
return "mdi:calendar"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["device_class"] = "date"
attributes["friendly_name"] = "Billing Current"
return attributes
>>>>>>> master
class NextBillDateSensor(FplDateEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Billing Next")
@property
def state(self):
<<<<<<< HEAD:custom_components/fpl1/sensor_DatesSensor.py
return datetime.date.fromisoformat(self.getData("next_bill_date"))
class ServiceDaysSensor(FplDayEntity):
=======
try:
self._state= datetime.date.fromisoformat(self.getData("next_bill_date"))
except:
pass
return self._state
<<<<<<< HEAD
class ServiceDaysSensor(FplDateEntity):
=======
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["device_class"] = "date"
attributes["friendly_name"] = "Billing Next"
return attributes
class ServiceDaysSensor(FplEntity):
>>>>>>> master
>>>>>>> master:custom_components/fpl/sensor_DatesSensor.py
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Billing Total Days")
@property
def state(self):
try:
self._state= self.getData("service_days")
except:
pass
return self._state
<<<<<<< HEAD
=======
@property
def icon(self):
return "mdi:calendar"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["unit_of_measurement"] = "days"
attributes["friendly_name"] = "Billing Total"
return attributes
>>>>>>> master
class AsOfDaysSensor(FplDayEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Billing As Of")
@property
def state(self):
try:
self._state= self.getData("as_of_days")
except:
pass
return self._state
<<<<<<< HEAD
=======
@property
def icon(self):
return "mdi:calendar"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["unit_of_measurement"] = "days"
attributes["friendly_name"] = "Billing As Of"
return attributes
>>>>>>> master
class RemainingDaysSensor(FplDayEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Billing Remaining")
@property
def state(self):
<<<<<<< HEAD
return self.getData("remaining_days")
=======
try:
self._state= self.getData("remaining_days")
except:
pass
return self._state
@property
def icon(self):
return "mdi:calendar"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["unit_of_measurement"] = "days"
attributes["friendly_name"] = "Billing Remaining"
return attributes
>>>>>>> master

196
custom_components/fpl1/sensor_KWHSensor.py
View File

@@ -1,196 +0,0 @@
"""energy sensors"""
from datetime import date, timedelta
import datetime
from homeassistant.components.sensor import (
STATE_CLASS_TOTAL_INCREASING,
STATE_CLASS_TOTAL,
)
from .fplEntity import FplEnergyEntity
class ProjectedKWHSensor(FplEnergyEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Projected")
@property
def state(self):
try:
self._state = self.getData("projectedKWH")
except:
pass
return self._state
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
<<<<<<< HEAD
class DailyAverageKWHSensor(FplEnergyEntity):
=======
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Projected KWH"
attributes["device_class"] = "energy"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "kWh"
return attributes
class DailyAverageKWHSensor(FplEntity):
>>>>>>> master
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Daily Average KWH")
@property
def state(self):
try:
self._state = self.getData("dailyAverageKWH")
except:
pass
return self._state
<<<<<<< HEAD
=======
@property
def icon(self):
return "mdi:flash"
<<<<<<< HEAD:custom_components/fpl1/sensor_KWHSensor.py
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
=======
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Daily Average"
attributes["device_class"] = "energy"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "kWh"
return attributes
>>>>>>> master
>>>>>>> master:custom_components/fpl/sensor_KWHSensor.py
class BillToDateKWHSensor(FplEnergyEntity):
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Bill To Date")
@property
def state(self):
try:
self._state = self.getData("billToDateKWH")
except:
pass
return self._state
def customAttributes(self):
"""Return the state attributes."""
# data = self.getData("daily_usage")
# date = data[-1]["readTime"]
asOfDays = self.getData("as_of_days")
last_reset = date.today() - timedelta(days=asOfDays)
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
attributes["last_reset"] = last_reset
return attributes
<<<<<<< HEAD
class NetReceivedKWHSensor(FplEnergyEntity):
=======
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Billing Usage"
attributes["device_class"] = "energy"
attributes["state_class"] = "total_increasing"
attributes["unit_of_measurement"] = "kWh"
if self.getData("billStartDate") is not None:
attributes["last_reset"] = self.getData("billStartDate")
return attributes
class NetReceivedKWHSensor(FplEntity):
>>>>>>> master
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Received Reading")
@property
def state(self):
try:
self._state = self.getData("recMtrReading")
except:
pass
return self._state
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
return attributes
<<<<<<< HEAD
class NetDeliveredKWHSensor(FplEnergyEntity):
=======
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Meter Return to Grid"
attributes["device_class"] = "energy"
attributes["state_class"] = "total_increasing"
attributes["unit_of_measurement"] = "kWh"
if self.getData("billStartDate") is not None:
attributes["last_reset"] = self.getData("billStartDate")
return attributes
class NetDeliveredKWHSensor(FplEntity):
>>>>>>> master
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Delivered Reading")
@property
def state(self):
try:
self._state = self.getData("delMtrReading")
except:
try:
self._state = self.getData("billToDateKWH")
except:
pass
return self._state
<<<<<<< HEAD:custom_components/fpl1/sensor_KWHSensor.py
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL_INCREASING
=======
@property
def icon(self):
return "mdi:flash"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Meter Consumption"
attributes["device_class"] = "energy"
attributes["state_class"] = "total_increasing"
attributes["unit_of_measurement"] = "kWh"
if self.getData("billStartDate") is not None:
attributes["last_reset"] = self.getData("billStartDate")
>>>>>>> master:custom_components/fpl/sensor_KWHSensor.py
return attributes

188
custom_components/fpl1/sensor_ProjectedBillSensor.py
View File

@@ -1,188 +0,0 @@
"""Projected bill sensors"""
from homeassistant.components.sensor import (
# STATE_CLASS_TOTAL_INCREASING,
STATE_CLASS_TOTAL,
)
from .fplEntity import FplMoneyEntity
class FplProjectedBillSensor(FplMoneyEntity):
"""projected bill sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Projected Bill")
@property
def native_value(self):
budget = self.getData("budget_bill")
budget_billing_projected_bill = self.getData("budget_billing_projected_bill")
if budget and budget_billing_projected_bill is not None:
return self.getData("budget_billing_projected_bill")
return self.getData("projected_bill")
"""
@property
def state(self):
budget = self.getData("budget_bill")
budget_billing_projected_bill = self.getData("budget_billing_projected_bill")
<<<<<<< HEAD:custom_components/fpl1/sensor_ProjectedBillSensor.py
if budget and budget_billing_projected_bill is not None:
return self.getData("budget_billing_projected_bill")
return self.getData("projected_bill")
"""
=======
try:
if budget == True and budget_billing_projected_bill is not None:
self._state = self.getData("budget_billing_projected_bill")
else:
self._state = self.getData("projected_bill")
except:
self._state = None
return self._state
>>>>>>> master:custom_components/fpl/sensor_ProjectedBillSensor.py
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
<<<<<<< HEAD:custom_components/fpl1/sensor_ProjectedBillSensor.py
attributes["state_class"] = STATE_CLASS_TOTAL
attributes["budget_bill"] = self.getData("budget_bill")
=======
attributes["friendly_name"] = "Projected Bill Payment Due"
attributes["device_class"] = "monetary"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "$"
>>>>>>> master:custom_components/fpl/sensor_ProjectedBillSensor.py
return attributes
# Defered Amount
class DeferedAmountSensor(FplMoneyEntity):
"""Defered amount sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Deferred Amount")
@property
def state(self):
<<<<<<< HEAD:custom_components/fpl1/sensor_ProjectedBillSensor.py
if self.getData("budget_bill"):
return self.getData("defered_amount")
return 0
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
=======
try:
self._state = self.getData("deferred_amount")
except:
self._state = 0
pass
return self._state
<<<<<<< HEAD
=======
@property
def icon(self):
return "mdi:currency-usd"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Deferred Amount"
attributes["device_class"] = "monetary"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "$"
return attributes
>>>>>>> master
>>>>>>> master:custom_components/fpl/sensor_ProjectedBillSensor.py
class ProjectedBudgetBillSensor(FplMoneyEntity):
"""projected budget bill sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Projected Budget Bill")
@property
def state(self):
try:
self._state = self.getData("budget_billing_projected_bill")
except:
pass
return self._state
<<<<<<< HEAD
=======
@property
def icon(self):
return "mdi:currency-usd"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Projected Budget Bill"
attributes["device_class"] = "monitary"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "$"
return attributes
<<<<<<< HEAD:custom_components/fpl1/sensor_ProjectedBillSensor.py
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
=======
>>>>>>> master
>>>>>>> master:custom_components/fpl/sensor_ProjectedBillSensor.py
class ProjectedActualBillSensor(FplMoneyEntity):
"""projeted actual bill sensor"""
def __init__(self, coordinator, config, account):
super().__init__(coordinator, config, account, "Projected Actual Bill")
@property
def state(self):
<<<<<<< HEAD
return self.getData("projected_bill")
<<<<<<< HEAD:custom_components/fpl1/sensor_ProjectedBillSensor.py
def customAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["state_class"] = STATE_CLASS_TOTAL
return attributes
=======
=======
try:
self._state = self.getData("projected_bill")
except:
pass
return self._state
@property
def icon(self):
return "mdi:currency-usd"
def defineAttributes(self):
"""Return the state attributes."""
attributes = {}
attributes["friendly_name"] = "Projected Actual Bill"
attributes["device_class"] = "monitary"
attributes["state_class"] = "total"
attributes["unit_of_measurement"] = "$"
return attributes
>>>>>>> master
>>>>>>> master:custom_components/fpl/sensor_ProjectedBillSensor.py