Skip to content

Commit

Permalink
Merge pull request #28 from faanskit/dev
Browse files Browse the repository at this point in the history 
Added kill switch and X-header
  • Loading branch information
@faanskit
faanskit authored Jan 19, 2024
2 parents 398cc02 + 6346d25 commit 321aa13
Show file tree
Hide file tree
Showing 6 changed files with 128 additions and 42 deletions.
6 changes: 6 additions & 0 deletions .vscode/settings.json
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
{
"editor.formatOnSave": true,
"python.formatting.provider": "black",
"python.linting.enabled": true,
"python.linting.flake8Enabled": true
}
6 changes: 3 additions & 3 deletions examples/main.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,10 @@ async def main(show_details=False):
# Create the async class
async with CheckwattManager(username, password) as check_watt_instance:
try:
# Login to EnergyInBalance
# Login to EnergyInBalance and check kill switch
if await check_watt_instance.login():
# Fetch customer detail
customer_id = await check_watt_instance.get_customer_details()
await check_watt_instance.get_customer_details()

# Do a sample
print("Customer Details\n================")
Expand Down Expand Up @@ -155,7 +155,7 @@ async def main(show_details=False):
print(f"Solar: {check_watt_instance.total_solar_energy/1000} kWh")
print(f"Charging: {check_watt_instance.total_charging_energy/1000} kWh")
print(
f"Discharging: {check_watt_instance.total_discharging_energy/1000} kWh"
f"Discharging: {check_watt_instance.total_discharging_energy/1000} kWh" # noqa: E501
)
print(f"Import: {check_watt_instance.total_import_energy/1000} kWh")
print(f"Export: {check_watt_instance.total_export_energy/1000} kWh")
Expand Down
147 changes: 111 additions & 36 deletions pycheckwatt/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
class CheckwattManager:
"""CheckWatt manager."""

def __init__(self, username, password) -> None:
def __init__(self, username, password, application="pyCheckwatt") -> None:
"""Initialize the CheckWatt manager."""
if username is None or password is None:
raise ValueError("Username and password must be provided.")
Expand Down Expand Up @@ -48,6 +48,7 @@ def __init__(self, username, password) -> None:
self.price_zone = None
self.spot_prices = None
self.energy_data = None
self.header_identifier = application

async def __aenter__(self):
"""Asynchronous enter."""
Expand All @@ -65,14 +66,15 @@ def _get_headers(self):
"accept": "application/json, text/plain, */*",
"accept-language": "sv-SE,sv;q=0.9,en-SE;q=0.8,en;q=0.7,en-US;q=0.6",
"content-type": "application/json",
"sec-ch-ua": '"Chromium";v="112", "Google Chrome";v="112", "Not:A-Brand";v="99"',
"sec-ch-ua": '"Chromium";v="112", "Google Chrome";v="112", "Not:A-Brand";v="99"', # noqa: E501
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Windows"',
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "cross-site",
"wslog-os": "",
"wslog-platform": "controlpanel",
"X-pyCheckwatt-Application": self.header_identifier,
}

def _extract_content_and_logbook(self, input_string):
Expand All @@ -95,7 +97,8 @@ def _extract_content_and_logbook(self, input_string):
# Extract logbook entries
logbook_entries = input_string.split("\n")

# Filter out entries containing #BEGIN_BATTERY_REGISTRATION and #END_BATTERY_REGISTRATION
# Filter out entries containing
# #BEGIN_BATTERY_REGISTRATION and #END_BATTERY_REGISTRATION
logbook_entries = [
entry.strip()
for entry in logbook_entries
Expand All @@ -109,7 +112,7 @@ def _extract_content_and_logbook(self, input_string):

def _extract_fcr_d_state(self):
pattern = re.compile(
r"\[ FCR-D (ACTIVATED|DEACTIVATE) \].*?(\d+,\d+/\d+,\d+/\d+,\d+ %).*?(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})"
r"\[ FCR-D (ACTIVATED|DEACTIVATE) \].*?(\d+,\d+/\d+,\d+/\d+,\d+ %).*?(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})" # noqa: E501
)
for entry in self.logbook_entries:
match = pattern.search(entry)
Expand All @@ -135,17 +138,59 @@ async def handle_client_error(self, endpoint, headers, error):
)
return False

async def _continue_kill_switch_not_enabled(self):
"""Check if CheckWatt has requested integrations to back-off."""
try:
url = "https://checkwatt.se/ha-killswitch.txt"
headers = {**self._get_headers()}
async with self.session.get(url, headers=headers) as response:
data = await response.text()
if response.status == 200:
kill = data.strip() # Remove leading and trailing whitespaces
if kill == "0":
# We are OK to continue
_LOGGER.debug(
"CheckWatt accepted and not enabled the kill-switch"
)
return True

# Kill was requested
_LOGGER.error(
"CheckWatt has requested to back down by enabling the kill-switch" # noqa: E501
)
return False

if response.status == 401:
_LOGGER.error(
"Unauthorized: Check your CheckWatt authentication credentials"
)
return False

_LOGGER.error("Unexpected HTTP status code: %s", response.status)
return False

except (ClientResponseError, ClientError) as error:
return await self.handle_client_error(url, headers, error)

async def login(self):
"""Login to CheckWatt."""
try:
if not await self._continue_kill_switch_not_enabled():
# CheckWatt want us to back down.
return False
_LOGGER.debug("Kill-switch not enabled, continue")

credentials = f"{self.username}:{self.password}"
encoded_credentials = base64.b64encode(credentials.encode("utf-8")).decode(
"utf-8"
)
endpoint = "/user/LoginEiB?audience=eib"

# Define headers with the encoded credentials
headers = {**self._get_headers(), "authorization": f"Basic {encoded_credentials}"}
headers = {
**self._get_headers(),
"authorization": f"Basic {encoded_credentials}",
}

async with self.session.get(
self.base_url + endpoint, headers=headers
Expand All @@ -157,7 +202,9 @@ async def login(self):
return True

if response.status == 401:
_LOGGER.error("Unauthorized: Check your checkwatt authentication credentials")
_LOGGER.error(
"Unauthorized: Check your checkwatt authentication credentials"
)
return False

_LOGGER.error("Unexpected HTTP status code: %s", response.status)
Expand All @@ -172,9 +219,14 @@ async def get_customer_details(self):
endpoint = "/controlpanel/CustomerDetail"

# Define headers with the JwtToken
headers = {**self._get_headers(), "authorization": f"Bearer {self.jwt_token}"}
headers = {
**self._get_headers(),
"authorization": f"Bearer {self.jwt_token}",
}

async with self.session.get(self.base_url + endpoint, headers=headers) as response:
async with self.session.get(
self.base_url + endpoint, headers=headers
) as response:
response.raise_for_status()
if response.status == 200:
self.customer_details = await response.json()
Expand Down Expand Up @@ -353,34 +405,49 @@ async def get_fcrd_revenueyear(self):
months = ["-01-01", "-06-30", "-07-01", yesterday_date]
loop = 0
retval = False
if (yesterday_date <= "-07-01"):
if yesterday_date <= "-07-01":
try:
year_date = datetime.now().strftime("%Y")
to_date = year_date + yesterday_date
from_date = year_date + "-01-01"
endpoint = f"/ems/fcrd/revenue?fromDate={from_date}&toDate={to_date}"
# Define headers with the JwtToken
headers = {**self._get_headers(),"authorization": f"Bearer {self.jwt_token}"}
headers = {
**self._get_headers(),
"authorization": f"Bearer {self.jwt_token}",
}
# First fetch the revenue
async with self.session.get(self.base_url + endpoint, headers=headers) as responseyear:
async with self.session.get(
self.base_url + endpoint, headers=headers
) as responseyear: # noqa: E501
responseyear.raise_for_status()
self.revenueyear = await responseyear.json()
for each in self.revenueyear:
self.revenueyeartotal += each["Revenue"]
if responseyear.status == 200:
# Then fetch the service fees
endpoint = (f"/ems/service/fees?fromDate={from_date}&toDate={to_date}")
async with self.session.get(self.base_url + endpoint, headers=headers) as responseyear:
endpoint = f"/ems/service/fees?fromDate={from_date}&toDate={to_date}" # noqa: E501
async with self.session.get(
self.base_url + endpoint, headers=headers
) as responseyear: # noqa: E501
responseyear.raise_for_status()
self.feesyear = await responseyear.json()
for each in self.feesyear["FCRD"]:
self.feesyeartotal += each["Revenue"]
if responseyear.status == 200:
retval = True
else:
_LOGGER.error("Obtaining data from URL %s failed with status code %d", self.base_url + endpoint, responseyear.status)
_LOGGER.error(
"Obtaining data from URL %s failed with status code %d", # noqa: E501
self.base_url + endpoint,
responseyear.status,
)
else:
_LOGGER.error("Obtaining data from URL %s failed with status code %d", self.base_url + endpoint, responseyear.status)
_LOGGER.error(
"Obtaining data from URL %s failed with status code %d",
self.base_url + endpoint,
responseyear.status,
)
return retval

except (ClientResponseError, ClientError) as error:
Expand All @@ -389,21 +456,28 @@ async def get_fcrd_revenueyear(self):
try:
while loop < 3:
year_date = datetime.now().strftime("%Y")
to_date = year_date + months[loop+1]
from_date = year_date + months[loop]
endpoint = f"/ems/fcrd/revenue?fromDate={from_date}&toDate={to_date}"
to_date = year_date + months[loop + 1]
from_date = year_date + months[loop]
endpoint = f"/ems/fcrd/revenue?fromDate={from_date}&toDate={to_date}" # noqa: E501
# Define headers with the JwtToken
headers = {**self._get_headers(),"authorization": f"Bearer {self.jwt_token}"}
headers = {
**self._get_headers(),
"authorization": f"Bearer {self.jwt_token}",
}
# First fetch the revenue
async with self.session.get(self.base_url + endpoint, headers=headers) as responseyear:
async with self.session.get(
self.base_url + endpoint, headers=headers
) as responseyear: # noqa: E501
responseyear.raise_for_status()
self.revenueyear = await responseyear.json()
for each in self.revenueyear:
self.revenueyeartotal += each["Revenue"]
if responseyear.status == 200:
# Then fetch the service fees
endpoint = (f"/ems/service/fees?fromDate={from_date}&toDate={to_date}")
async with self.session.get(self.base_url + endpoint, headers=headers) as responseyear:
endpoint = f"/ems/service/fees?fromDate={from_date}&toDate={to_date}" # noqa: E501
async with self.session.get(
self.base_url + endpoint, headers=headers
) as responseyear:
responseyear.raise_for_status()
self.feesyear = await responseyear.json()
for each in self.feesyear["FCRD"]:
Expand All @@ -412,9 +486,17 @@ async def get_fcrd_revenueyear(self):
loop += 2
retval = True
else:
_LOGGER.error("Obtaining data from URL %s failed with status code %d", self.base_url + endpoint, responseyear.status)
_LOGGER.error(
"Obtaining data from URL %s failed with status code %d", # noqa: E501
self.base_url + endpoint,
responseyear.status,
)
else:
_LOGGER.error("Obtaining data from URL %s failed with status code %d", self.base_url + endpoint, responseyear.status)
_LOGGER.error(
"Obtaining data from URL %s failed with status code %d", # noqa: E501
self.base_url + endpoint,
responseyear.status,
)
return retval

except (ClientResponseError, ClientError) as error:
Expand Down Expand Up @@ -500,7 +582,6 @@ async def get_energy_flow(self):
except (ClientResponseError, ClientError) as error:
return await self.handle_client_error(endpoint, headers, error)


async def get_price_zone(self):
"""Fetch Price Zone from checkwatt."""

Expand Down Expand Up @@ -540,7 +621,7 @@ async def get_spot_price(self):
to_date = end_date.strftime("%Y-%m-%d")
if self.price_zone is None:
await self.get_price_zone()
endpoint = f"/ems/spotprice?zone={self.price_zone}&fromDate={from_date}&toDate={to_date}"
endpoint = f"/ems/spotprice?zone={self.price_zone}&fromDate={from_date}&toDate={to_date}" # noqa: E501
# Define headers with the JwtToken
headers = {
**self._get_headers(),
Expand All @@ -566,7 +647,6 @@ async def get_spot_price(self):
except (ClientResponseError, ClientError) as error:
return await self.handle_client_error(endpoint, headers, error)


async def get_energy_trading_company(self, input_id):
"""Translate Energy Company Id to Energy Company Name."""
try:
Expand All @@ -584,9 +664,8 @@ async def get_energy_trading_company(self, input_id):
if response.status == 200:
energy_trading_companies = await response.json()
for energy_trading_company in energy_trading_companies:
if energy_trading_company['Id'] == input_id:
return energy_trading_company['DisplayName']

if energy_trading_company["Id"] == input_id:
return energy_trading_company["DisplayName"]

return None

Expand All @@ -600,8 +679,6 @@ async def get_energy_trading_company(self, input_id):
except (ClientResponseError, ClientError) as error:
return await self.handle_client_error(endpoint, headers, error)



@property
def inverter_make_and_model(self):
"""Property for inverter make and model. Not used by HA integration.."""
Expand All @@ -622,7 +699,7 @@ def battery_make_and_model(self):
):
resp = f"{self.battery_registration['BatterySystem']}"
resp += f" {self.battery_registration['BatteryModel']}"
resp += f" ({self.battery_registration['BatteryPowerKW']}kW, {self.battery_registration['BatteryCapacityKWh']}kWh)"
resp += f" ({self.battery_registration['BatteryPowerKW']}kW, {self.battery_registration['BatteryCapacityKWh']}kWh)" # noqa: E501
return resp
else:
return "Could not get any information about your battery"
Expand All @@ -637,7 +714,7 @@ def electricity_provider(self):
resp = f"{self.battery_registration['ElectricityCompany']}"
resp += f" via {self.battery_registration['Dso']}"
if "GridAreaId" in self.battery_registration:
resp += f" ({self.battery_registration['GridAreaId']} {self.battery_registration['Kommun']})"
resp += f" ({self.battery_registration['GridAreaId']} {self.battery_registration['Kommun']})" # noqa: E501
return resp

@property
Expand Down Expand Up @@ -772,7 +849,6 @@ def get_spot_price_excl_vat(self, now_hour: int):
_LOGGER.warning("Unable to retrieve spot price for the current hour")
return None


@property
def battery_power(self):
"""Property for Battery Power."""
Expand Down Expand Up @@ -812,4 +888,3 @@ def battery_soc(self):

_LOGGER.warning("Unable to retrieve Battery SoC")
return None

2 changes: 1 addition & 1 deletion pyproject.toml
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "pycheckwatt"
version = "0.1.8"
version = "0.1.9"
description = "Read data from CheckWatts EnergyInBalance WEB API"
authors = ["Marcus Karlsson <[email protected]>", "Anders Yderborg <[email protected]>", "Daniel Nilsson <[email protected]>"]
license = "MIT License"
Expand Down
7 changes: 6 additions & 1 deletion setup.cfg
View file
Original file line number Diff line number Diff line change
@@ -1,2 +1,7 @@
[metadata]
description-file = README.md
description-file = README.md
[flake8]
max-line-length = 88
exclude = .git,__pycache__,venv
[tool.black]
line-length = 88
2 changes: 1 addition & 1 deletion setup.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@

MIN_PY_VERSION = "3.10"
PACKAGES = find_packages()
VERSION = "0.1.8"
VERSION = "0.1.9"

setup(
name="pycheckwatt",
Expand Down

0 comments on commit 321aa13

Please sign in to comment.