wip: Added reading of powerstation sensor.

Author: StephanU

custom_components/talent_monitor/const.py

@@ -13,7 +13,7 @@
 
 # Platforms
 SENSOR = "sensor"
-PLATFORMS = []
+PLATFORMS = [SENSOR]
 
 
 # Configuration and options

custom_components/talent_monitor/coordinator.py

@@ -52,7 +52,7 @@ async def _async_update_data(self):
         """Update data via library."""
         _LOGGER.debug("_async_update_data ")
         try:
-            return await self.api.fetch_solar_data()
+            return await self.api.fetch_data()
         except Exception as exception:
             _LOGGER.exception("_async_update_data failed")
             raise UpdateFailed() from exception

custom_components/talent_monitor/entity.py

@@ -0,0 +1,36 @@
+"""TalentMonitorEntity class."""
+import logging
+
+from custom_components.talent_monitor.pyTalentMonitor.data_provider import Entity
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .const import NAME
+
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+
+class TalentMonitorEntity(CoordinatorEntity):
+    """Base Class for TalentMonitor entities."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self, coordinator, entity: Entity
+    ):
+        """Initialize a TalentMonitor entity."""
+        super().__init__(coordinator)
+
+        device_id = f"{entity.entity_id}"
+        device_name = entity.name
+
+        self._attr_unique_id = f"{device_id}"
+
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device_id)},
+            manufacturer=NAME,
+            name=device_name,
+        )
+
+        _LOGGER.debug("Added TalentMonitor entity id='%s'", self.unique_id)

custom_components/talent_monitor/pyTalentMonitor/__init__.py

@@ -4,23 +4,22 @@
 import asyncio
 import json
 import logging
-import os
 
 from aiohttp import ClientSession
+from custom_components.talent_monitor.pyTalentMonitor.inverter import InverterDataProvider
+from custom_components.talent_monitor.pyTalentMonitor.data_provider import DataProvider
+from custom_components.talent_monitor.pyTalentMonitor.power_station import PowerStation, PowerStationDataProvider
 
 # Configure logging
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
 BASE_URL = "https://www.talent-monitoring.com/prod-api"
 TIMEZONE = "+02:00"
 
-
 class AuthenticationError(Exception):
     """AuthenticationError when connecting to the Talent API."""
-
     pass
 
-
 class TalentSolarMonitor:
     """TalentSolarMonitor API client."""
 
@@ -29,125 +28,27 @@ def __init__(
         username: str = None,
         password: str = None,
         session: ClientSession = None,
-        return_json: bool = False,
     ):
         """Construct the TalentSolarMonitor API client."""
-        self.username = username or os.environ.get("PYTALENT_USERNAME")
-        self.password = password or os.environ.get("PYTALENT_PASSWORD")
-        self.session = session
-        self.return_json = return_json
-        self.token = None
-
-    def get_credentials(self):
-        """Check whether the credentials are set."""
-        if not self.username or not self.password:
-            raise ValueError(
-                "Credentials not provided via command line arguments or environment variables."
-            )
+        self._data_provider = DataProvider(username, password, session)
+        self._inverter_data_provider = InverterDataProvider(self._data_provider)
+        self._power_station_data_provider = PowerStationDataProvider(self._data_provider)
 
-    async def login(self):
-        """Log in using the given credentials."""
-        login_data = {"username": self.username, "password": self.password}
-        response = await self.session.post(f"{BASE_URL}/login", json=login_data)
-        response_data = await response.json()
-        if "token" in response_data:
-            self.token = response_data["token"]
-            _LOGGER.debug("Login successful - received token: %s", self.token)
-        else:
-            _LOGGER.error("Login failed. Got status code %s", response.status)
-            raise AuthenticationError("Authentication failed")
 
-    async def refresh_token(self):
-        """Refresh the token."""
-        _LOGGER.debug("Token expired. Refreshing token...")
-        self.login()
+    def get_power_stations(self) -> list[PowerStation]:
+        return self._power_station_data_provider.power_stations
 
-    async def get_data(self, endpoint):
-        """Get data from the given endpoint."""
-        if not self.token:
-            self.login()
-        headers = {"Authorization": f"Bearer {self.token}"}
-        response = await self.session.get(f"{BASE_URL}/{endpoint}", headers=headers)
-        if response.status == 401:  # Unauthorized, token might be expired
-            self.refresh_token()
-            headers["Authorization"] = f"Bearer {self.token}"
-            response = await self.session.get(f"{BASE_URL}/{endpoint}", headers=headers)
-
-        if response.status == 200:
-            return await response.json()
-        else:
-            _LOGGER.error("Failed to fetch data. Status Code: %s", response.status)
-            return None
+    def fetch_data(self):
+        self._inverter_data_provider.fetch_data()
+        self._power_station_data_provider.fetch_data()
 
     async def fetch_solar_data(self):
-        """Fetch the solar data."""
-        self.get_credentials()
-        await self.login()
-
-        data = await self.get_data(endpoint="system/station/list")
-        if data and "rows" in data and len(data["rows"]) > 0:
-            first_station = data["rows"][0]
-            status = first_station["status"]
-            stationName = first_station["stationName"]
-            powerStationGuid = first_station["powerStationGuid"]
-            _LOGGER.debug("GUID: %s", powerStationGuid)
-
-            data = await self.get_data(
-                endpoint=f"system/station/getPowerStationByGuid?powerStationGuid={powerStationGuid}&timezone={TIMEZONE}"
-            )
-            _LOGGER.debug("Data for powerstation GUID %s: %s", powerStationGuid, json.dumps(data))
-
-            if data:
-                power_data = data["data"]
-                totalActivePower = power_data["totalActivePower"]
-                dayEnergy = power_data["dayEnergy"]
-                monthEnergy = power_data["monthEnergy"]
-                yearEnergy = power_data["yearEnergy"]
+        await self.fetch_data()
 
-            data = await self.get_data(endpoint="tools/device/selectDeviceInverter")
-            if data:
-                deviceGuid = data["rows"][0]["deviceGuid"]
-
-            data = await self.get_data(
-                endpoint=f"tools/device/selectDeviceInverterInfo?deviceGuid={deviceGuid}"
-            )
-
-            _LOGGER.debug("Data for inverter GUID %s: %s", deviceGuid, json.dumps(data))
-            if data:
-                pv = data["data"]["pv"]
-                pv1Voltage = pv[0]["voltage"]
-                pv1Current = pv[0]["current"]
-                pv1Power = pv[0]["power"]
-                pv2Voltage = pv[1]["voltage"]
-                pv2Current = pv[1]["current"]
-                pv2Power = pv[1]["power"]
-
-            result = {
-                "Status": status,
-                "StationName": stationName,
-                "TotalActivePower(W)": totalActivePower,
-                "DailyEnergy(Wh)": dayEnergy,
-                "MonthlyEnergy(Wh)": monthEnergy,
-                "YearlyEnergy(Wh)": yearEnergy,
-                "Panel1Voltage(V)": pv1Voltage,
-                "Panel1Current(A)": pv1Current,
-                "Panel1Power(W)": pv1Power,
-                "Panel2Voltage(V)": pv2Voltage,
-                "Panel2Current(A)": pv2Current,
-                "Panel2Power(W)": pv2Power,
-            }
-
-            if self.return_json:
-                return json.dumps(result, indent=4)
-            else:
-                for key, value in result.items():
-                    _LOGGER.debug("%s: %s",key, value)
-
-
-async def main(username: str, password: str, return_json: bool):
+async def main(username: str, password: str):
     """Connect to the TalentSolarMonitor API and fetch the solar data."""
     async with ClientSession() as session:
-        talent_monitor = TalentSolarMonitor(username, password, session, return_json)
+        talent_monitor = TalentSolarMonitor(username, password, session)
         result = await talent_monitor.fetch_solar_data()
         if result:
             _LOGGER.info("Solar data received: %s", result)
@@ -159,9 +60,6 @@ async def main(username: str, password: str, return_json: bool):
     )
     parser.add_argument("-u", "--username", required=False, help="Username to log in")
     parser.add_argument("-p", "--password", required=False, help="Password to log in")
-    parser.add_argument(
-        "--json", action="store_true", help="Return output as JSON object"
-    )
     args = parser.parse_args()
 
-    asyncio.run(main(args.username, args.password, args.json))
+    asyncio.run(main(args.username, args.password))

custom_components/talent_monitor/pyTalentMonitor/data_provider.py

@@ -0,0 +1,70 @@
+import logging
+import os
+
+from aiohttp import ClientSession
+from custom_components.talent_monitor.pyTalentMonitor import AuthenticationError
+
+# Configure logging
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+BASE_URL = "https://www.talent-monitoring.com/prod-api"
+
+class DataProvider():
+    """Data provider accessing the MyGekko API"""
+
+    def __init__(
+        self,  username: str, password: str, session: ClientSession
+    ):
+        self._url = BASE_URL
+        self._username = username or os.environ.get("PYTALENT_USERNAME")
+        self._password = password or os.environ.get("PYTALENT_PASSWORD")
+        self._session = session
+        self._token = None
+
+    def get_credentials(self):
+        """Check whether the credentials are set."""
+        if not self._username or not self.password:
+            raise ValueError(
+                "Credentials not provided via command line arguments or environment variables."
+            )
+
+    async def login(self):
+        """Log in using the given credentials."""
+        login_data = {"username": self._username, "password": self._password}
+        response = await self.session.post(f"{self._url}/login", json=login_data)
+        response_data = await response.json()
+        if "token" in response_data:
+            self._token = response_data["token"]
+            _LOGGER.debug("Login successful - received token: %s", self._token)
+        else:
+            _LOGGER.error("Login failed. Got status code %s", response.status)
+            raise AuthenticationError("Authentication failed")
+
+    async def refresh_token(self):
+        """Refresh the token."""
+        _LOGGER.debug("Token expired. Refreshing token...")
+        self.login()
+
+    async def get_data(self, endpoint):
+        """Get data from the given endpoint."""
+        if not self._token:
+            self.login()
+        headers = {"Authorization": f"Bearer {self._token}"}
+        response = await self.session.get(f"{self._url}/{endpoint}", headers=headers)
+        if response.status == 401:  # Unauthorized, token might be expired
+            self.refresh_token()
+            headers["Authorization"] = f"Bearer {self._token}"
+            response = await self.session.get(f"{self._url}/{endpoint}", headers=headers)
+
+        if response.status == 200:
+            return await response.json()
+        else:
+            _LOGGER.error("Failed to fetch data. Status Code: %s", response.status)
+            return None
+
+class Entity:
+    """Base class for TalentMonitor entities"""
+
+    def __init__(self, entity_id: str, name: str) -> None:
+        self.entity_id = entity_id
+        self.name = name

custom_components/talent_monitor/pyTalentMonitor/inverter.py

@@ -0,0 +1,61 @@
+"""TalentMonitor Inverter"""
+
+import json
+import logging
+
+from custom_components.talent_monitor.pyTalentMonitor.data_provider import DataProvider, Entity
+
+# Configure logging
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+class Inverter(Entity):
+    def __init__(
+        self, entity_id: str, name: str
+    ) -> None:
+        super().__init__(entity_id, name)
+        self._data
+
+    @property
+    def data(self, data):
+        self._data = data
+
+    @property
+    def data(self):
+        return self._data
+
+
+class InverterDataProvider():
+    def __init__(
+        self, data_provider: DataProvider
+    ) -> None:
+        self._data_provider = data_provider
+        self._inverter = {}
+
+    @property
+    def inverters(self) -> list[Inverter]:
+        """Returns the inverters read from TalentMonitor"""
+        result: list[Inverter] = []
+        for key, data in self._inverter.items():
+            result.append(data)
+
+        return result
+
+    async def fetch_data(self):
+        data = await self._data_provider.get_data(endpoint="tools/device/selectDeviceInverter")
+        if data and "rows" in data:
+            for inverter_data in data["rows"]:
+                if "deviceGuid" in inverter_data:
+                    deviceGuid = inverter_data["deviceGuid"]
+
+                    if not deviceGuid in self._inverters:
+                        self._inverters["deviceGuid"] = Inverter()
+
+                    inverter = self._inverters["deviceGuid"]
+
+                    inverter_info = await self._data_provider.get_data(
+                        endpoint=f"tools/device/selectDeviceInverterInfo?deviceGuid={deviceGuid}"
+                    )
+
+                    _LOGGER.debug("Data for inverter GUID %s: %s", deviceGuid, json.dumps(inverter_info))
+                    if inverter_info:
+                        inverter.data = inverter_info