Added kill switch and X-header

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+    "editor.formatOnSave": true,
+    "python.formatting.provider": "black",
+    "python.linting.enabled": true,
+    "python.linting.flake8Enabled": true
+}

examples/main.py

@@ -21,7 +21,7 @@ async def main(show_details=False):
             # Login to EnergyInBalance
             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================")
@@ -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")

pycheckwatt/__init__.py

@@ -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.")
@@ -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."""
@@ -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):
@@ -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
@@ -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)
@@ -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
@@ -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)
@@ -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()
@@ -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:
@@ -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"]:
@@ -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:
@@ -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."""
 
@@ -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(),
@@ -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:
@@ -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
 
@@ -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.."""
@@ -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"
@@ -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
@@ -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."""
@@ -812,4 +888,3 @@ def battery_soc(self):
 
         _LOGGER.warning("Unable to retrieve Battery SoC")
         return None
-

pyproject.toml

@@ -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"

setup.cfg

@@ -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

setup.py

@@ -3,7 +3,7 @@
 
 MIN_PY_VERSION = "3.10"
 PACKAGES = find_packages()
-VERSION = "0.1.8"
+VERSION = "0.1.9"
 
 setup(
     name="pycheckwatt",