Adding functions to find associated devices of a plant and inverters data.

src/fusion_solar_py/client.py

@@ -428,7 +428,7 @@ def _configure_session(self):
         )
 
         # the new API returns a 500 exception if the subdomain is incorrect
-        if r.status_code == 500:
+        if r.status_code in [500, 400]:
             try:
                 data = r.json()
 
@@ -573,6 +573,45 @@ def get_current_plant_data(self, plant_id: str) -> dict:
 
         return power_obj["data"]
 
+    @logged_in
+    def get_device_data(self, device_id: str, signal_ids: list[int] | list[str] = None) -> dict:
+        """
+        Retrieves measurements for the specified device and signals.
+        Each MPPT ID follow this logic :
+            Start at 11001. This ID corresponds to the voltage (V) of the first MPPT.
+            11002 corresponds to the intensity (A) of the first MPPT.
+            Add 2 to that (ends up being 11004) corresponds to the voltage (V) of the second MPPT.
+            11005 corresponds to the second MPPT's intensity (A)
+            And so on.
+        :param device_id: the device ID
+        :param signal_ids: signal ids to query
+
+        :return: a dict containing the asked data
+        """
+
+        if signal_ids is None:
+            # generates the 20 first MPPT IDs for voltage and intensity.
+            signal_ids = [val for i in range(10) for val in (11001 + i * 3, 11002 + i * 3)]
+
+        url = f"https://{self._huawei_subdomain}.fusionsolar.huawei.com/rest/pvms/web/device/v1/device-real-kpi"
+        params = {
+            "deviceDn": device_id,
+            "signalIds": signal_ids,
+            "_": round(time.time() * 1000),
+        }
+
+        r = self._session.get(url=url, params=params)
+        r.raise_for_status()
+
+        # errors in decoding the object generally mean that the login expired
+        # this is handeled by @logged_in
+        data_obj = r.json()
+
+        if "data" not in data_obj:
+            raise FusionSolarException("Failed to retrieve plant data.")
+
+        return data_obj["data"]
+
     @logged_in
     def get_plant_ids(self) -> list:
         """Get the ids of all available stations linked
@@ -621,12 +660,13 @@ def get_station_list(self) -> list:
 
 
     @logged_in
-    def get_device_ids(self) -> list:
-        """gets the devices associated to a given parent_id (can be a plant or a company/account)
+    def get_device_ids(self, parent_id: str = None) -> list:
+        """gets the devices associated to a given parent_id
+        :param parent_id: the parent_id (can be a plant or a company/account)
         returns a dictionary mapping device_type to device_id"""
         url = f"https://{self._huawei_subdomain}.fusionsolar.huawei.com/rest/neteco/web/config/device/v1/device-list"
         params = {
-            "conditionParams.parentDn": self._company_id,  # can be a plant or company id
+            "conditionParams.parentDn": self._company_id if parent_id is None else parent_id,  # can be a plant or company id
             "conditionParams.mocTypes": "20814,20815,20816,20819,20822,50017,60066,60014,60015,23037",  # specifies the types of devices
             "_": round(time.time() * 1000),
         }
@@ -638,6 +678,8 @@ def get_device_ids(self) -> list:
         for device in device_data["data"]:
             devices += [dict(type=device["mocTypeName"], deviceDn=device["dn"])]
         return devices
+
+
     @logged_in
     def get_historical_data(self, signal_ids: list[str] = ['30014', '30016', '30017'], device_dn:str = None, date: datetime = datetime.now() ) -> dict:
         """retrieves historical data for specified signals and device

tests/test_client.py

@@ -76,6 +76,20 @@ def test_current_plant_data(self):
         self.assertIsNotNone((current_plant_data))
         self.assertTrue(current_plant_data["yearEnergy"] > 0)
 
+    def test_get_device_data(self):
+        client = FusionSolarClient(self.user, self.password, self.subdomain)
+
+        device_ids = client.get_device_ids()
+        inverters = list(filter(lambda e: e['type'] == 'Inverter', device_ids))
+        inverter_id = inverters[0]['deviceDn']
+
+        device_data = client.get_device_data(inverter_id, [11001, 11002])
+
+        self.assertIsNotNone(device_data.get("signals"))
+        signals = device_data["signals"]
+        self.assertIsNotNone(signals.get("11001"))
+        self.assertIsNotNone(signals.get("11002"))
+
     def test_get_plant_stats(self):
         client = FusionSolarClient(self.user, self.password, self.subdomain)
 
@@ -111,6 +125,7 @@ def test_get_plant_stats(self):
 
         # get the device ids
         device_ids = client.get_device_ids()
+        plant_devices_ids = client.get_device_ids(plant_ids[0])
 
         self.assertTrue(len(device_ids) > 0)
         inverters = list(filter(lambda e: e['type'] == 'Inverter', device_ids))