Added class for CheckWattRank

examples/history_inject.py

@@ -1,8 +1,6 @@
 """Pulls historical data from CheckWatt EnergyInBalance and push it to CheckWattRank"""
 
-import aiohttp
-
-from pycheckwatt import CheckwattManager
+from pycheckwatt import CheckwattManager, CheckWattRankManager
 
 START_DATE = "2023-09-01"
 END_DATE = "2024-01-31"
@@ -45,36 +43,25 @@ async def main():
                     print("Failed to fetch electricity compan")
                     return
 
-                data = {
-                    "display_name": (
-                        DISPLAY_NAME_OVERRIDE
-                        if DISPLAY_NAME_OVERRIDE != ""
-                        else cw.display_name
-                    ),
-                    "dso": cw.battery_registration["Dso"],
-                    "electricity_area": cw.price_zone,
-                    "installed_power": cw.battery_charge_peak_ac,
-                    "electricity_company": energy_provider,
-                    "reseller_id": cw.reseller_id,
-                    "reporter": "CheckWattRank",
-                    "historical_data": hd,
-                }
-
-                # Post data to Netlify function
-                netlify_function_url = BASE_URL + "/.netlify/functions/publishHistory"
-                async with aiohttp.ClientSession() as session:
-                    async with session.post(
-                        netlify_function_url, json=data
-                    ) as response:
-                        if response.status == 200:
-                            result = await response.json()
-                            count = result.get("count", 0)
-                            total = result.get("total", 0)
-                            print(f"Data posted successfully. Count: {count}/{total}")
-                        else:
-                            print(
-                                f"Failed to post data. Status code: {response.status}"
-                            )
+                async with CheckWattRankManager() as cwr:
+                    (status, stored_items, total_items) = (
+                        await cwr.push_history_to_checkwatt_rank(
+                            display_name=(
+                                DISPLAY_NAME_OVERRIDE
+                                if DISPLAY_NAME_OVERRIDE != ""
+                                else cw.display_name
+                            ),
+                            dso=cw.battery_registration["Dso"],
+                            electricity_company=energy_provider,
+                            electricity_area=cw.price_zone,
+                            installed_power=cw.battery_charge_peak_ac,
+                            reseller_id=cw.reseller_id,
+                            reporter="CheckWattRank",
+                            historical_data=hd,
+                        )
+                    )
+                    print(f"Result: {status}")
+                    print(f"Count:  {stored_items}/{total_items}")
 
         except Exception as e:
             print(f"An error occurred: {e}")

pycheckwatt/__init__.py

@@ -1104,3 +1104,147 @@ def meter_version(self):
 
         _LOGGER.warning("Unable to find Meter Data for Meter Version")
         return None
+
+
+class CheckWattRankManager:
+    def __init__(self) -> None:
+        self.session = None
+        self.base_url = "https://checkwattrank.netlify.app"
+
+    async def __aenter__(self):
+        """Asynchronous enter."""
+        self.session = ClientSession()
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        """Asynchronous exit."""
+        await self.session.close()
+
+    async def push_to_checkwatt_rank(
+        self,
+        display_name,
+        dso,
+        electricity_company,
+        electricity_area,
+        installed_power,
+        today_net_income,
+        reseller_id,
+        reporter,
+    ):
+        """Push Data to CheckWattRank."""
+
+        headers = {
+            "Content-Type": "application/json",
+        }
+        url = self.base_url + "/.netlify/functions/publishToSheet"
+
+        payload = {
+            "display_name": display_name,
+            "dso": dso,
+            "electricity_company": electricity_company,
+            "electricity_area": electricity_area,
+            "installed_power": installed_power,
+            "today_net_income": today_net_income,
+            "reseller_id": reseller_id,
+            "reporter": reporter,
+        }
+
+        timeout_seconds = 10
+        async with ClientSession() as session:
+            try:
+                async with session.post(
+                    url, headers=headers, json=payload, timeout=timeout_seconds
+                ) as response:
+                    response.raise_for_status()
+                    content_type = response.headers.get("Content-Type", "").lower()
+                    _LOGGER.debug(
+                        "CheckWattRank Push Response Content-Type: %s",
+                        content_type,
+                    )
+
+                    if "application/json" in content_type:
+                        result = await response.json()
+                        _LOGGER.debug("CheckWattRank Push Response: %s", result)
+                        return True
+                    elif "text/plain" in content_type:
+                        result = await response.text()
+                        _LOGGER.debug("CheckWattRank Push Response: %s", result)
+                        return True
+                    else:
+                        _LOGGER.warning("Unexpected Content-Type: %s", content_type)
+                        result = await response.text()
+                        _LOGGER.debug("CheckWattRank Push Response: %s", result)
+
+            except ClientError as e:
+                _LOGGER.error("Error pushing data to CheckWattRank: %s", e)
+
+            except TimeoutError:
+                _LOGGER.error(
+                    "Request to CheckWattRank timed out after %s seconds",
+                    timeout_seconds,
+                )
+        return False
+
+    async def push_history_to_checkwatt_rank(
+        self,
+        display_name,
+        dso,
+        electricity_company,
+        electricity_area,
+        installed_power,
+        reseller_id,
+        reporter,
+        historical_data,
+    ):
+        headers = {
+            "Content-Type": "application/json",
+        }
+        url = self.base_url + "/.netlify/functions/publishHistory"
+
+        payload = {
+            "display_name": display_name,
+            "dso": dso,
+            "electricity_company": electricity_company,
+            "electricity_area": electricity_area,
+            "installed_power": installed_power,
+            "reseller_id": reseller_id,
+            "reporter": reporter,
+            "historical_data": historical_data,
+        }
+        timeout_seconds = 10
+        stored_items = 0
+        total_items = 0
+        status = None
+        async with ClientSession() as session:
+            try:
+                async with session.post(
+                    url, headers=headers, json=payload, timeout=timeout_seconds
+                ) as response:
+                    if response.status == 200:
+                        result = await response.json()
+                        stored_items = result.get("count", 0)
+                        total_items = result.get("total", 0)
+                        status = result.get("message", 0)
+                    else:
+                        _LOGGER.debug(
+                            "Failed to post data. Status code: %s",
+                            response.status,
+                        )
+                        status = f"Failed to post data. Status code: {response.status}"
+
+            except ClientError as e:
+                _LOGGER.error("Error pushing data to CheckWattRank: %s", e)
+                status = f"Failed to push historical data. Error {e}"
+
+            except TimeoutError:
+                _LOGGER.error(
+                    "Request to CheckWattRank timed out after %s seconds",
+                    timeout_seconds,
+                )
+                status = "Timeout pushing historical data."
+
+        return (
+            status,
+            stored_items,
+            total_items,
+        )

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pycheckwatt"
-version = "0.2.1"
+version = "0.2.2"
 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.py

@@ -4,7 +4,7 @@
 
 MIN_PY_VERSION = "3.10"
 PACKAGES = find_packages()
-VERSION = "0.2.1"
+VERSION = "0.2.2"
 
 setup(
     name="pycheckwatt",