Add JWT access to api

README.md

@@ -1,21 +1,25 @@
 # Fronius Solar.web
 
-Python client for the Fronius [Solar.web API](https://www.fronius.com/~/downloads/Solar%20Energy/User%20Information/SE_UI_API_InterfaceDocumentation_EN.pdf).
+Python client for the Fronius [Solar.web API documentation](https://www.fronius.com/~/downloads/Solar%20Energy/User%20Information/SE_UI_API_InterfaceDocumentation_EN.pdf). 
+Fronius [test site](https://api.solarweb.com/swqapi/index.html).
 
 ## Features 
 
 - Talks to your Fronius Solar.web PV system via Cloud API
 - Automatic retries with exponential backoff
 - Optionally pass in a `httpx` client
+- If a login and password is provided login with a bearer token can be used
 
 ## Usage
 
-Although intended as a library [`fronius_sw_example.py`](https://github.com/drc38/python-fronius-web/blob/main/examples/fronius_sw_example.py) is provided for testing purposes.
+Although intended as a library [`fronius_key_example.py`](https://github.com/drc38/python-fronius-web/blob/main/examples/fronius_key_example.py) is provided for testing purposes.
 
 Authentication and PV system id for the example is provided via environment variables, e.g. on nix systems:
 
 ```
 export ACCESS_KEY_ID=FKIAFEF58CFEFA94486F9C804CF6077A01AB
 export ACCESS_KEY_VALUE=47c076bc-23e5-4949-37a6-4bcfcf8d21d6
+export LOGIN_NAME=abc@email.com
+export LOGIN_PASSWORD=xxxxx
 export PV_SYSTEM_ID=20bb600e-019b-4e03-9df3-a0a900cda689
 ```

examples/fronius_login_example.py

@@ -0,0 +1,52 @@
+import asyncio
+from pydantic import SecretStr
+from pydantic_settings import BaseSettings
+from fronius_solarweb.api import Fronius_Solarweb
+
+
+class AuthDetails(BaseSettings):
+    ACCESS_KEY_ID: SecretStr
+    ACCESS_KEY_VALUE: SecretStr
+    LOGIN_NAME: SecretStr
+    LOGIN_PASSWORD: SecretStr
+    PV_SYSTEM_ID: str
+
+
+async def main():
+    creds = AuthDetails()
+    fronius = Fronius_Solarweb(
+        access_key_id=creds.ACCESS_KEY_ID.get_secret_value(),
+        access_key_value=creds.ACCESS_KEY_VALUE.get_secret_value(),
+        login_name=creds.LOGIN_NAME.get_secret_value(),
+        login_password=creds.LOGIN_PASSWORD.get_secret_value(),
+        pv_system_id=creds.PV_SYSTEM_ID,
+    )
+    await fronius.login()
+
+    print("Getting SolarWeb api release info:")
+    release_info = await fronius.get_api_release_info()
+    print(f"{release_info}\n")
+
+    print("Getting PV systems meta data:")
+    pv_systems_data = await fronius.get_pvsystems_meta_data()
+    print(f"{pv_systems_data}\n")
+
+    print(f"Getting PV system meta data for {creds.PV_SYSTEM_ID}:")
+    pv_system_data = await fronius.get_pvsystem_meta_data()
+    print(f"{pv_system_data}\n")
+
+    print("Getting Devices meta data:")
+    devices_data = await fronius.get_devices_meta_data()
+    print(f"{devices_data}\n")
+
+    print("Getting power flow data:")
+    flow_data = await fronius.get_system_flow_data()
+    print(f"{flow_data}\n")
+
+    print("Getting aggregated V2 data:")
+    aggr_data = await fronius.get_system_aggr_data_v2()
+    print(f"{aggr_data}\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

fronius_solarweb/api.py

@@ -31,50 +31,107 @@
 class Fronius_Solarweb:
     def __init__(
         self,
-        access_key_id: str,
-        access_key_value: str,
+        access_key_id: str = None,
+        access_key_value: str = None,
+        login_name: str = None,
+        login_password: str = None,
         pv_system_id: str = None,
         httpx_client: AsyncClient = None,
     ):
         """
-        Create an Fronius Solarweb API client.
+        Create an Fronius Solarweb API client from either key/id or login/password.
 
         :param access_key_id:  unique ID for the API key, e.g.
             "FKIAFEF58CFEFA94486F9C804CF6077A01AB". Access keys are 36
             characters long and start with the "FKIA" prefix.
         :param access_key_value: A secret value (GUID),
             e.g. "47c076bc-23e5-4949-37a6-4bcfcf8d21d6", which
             you need to know for authorization of API calls.
+            :param login_name: Solar.web email / login name.
+            :param login_password: Solar.web password.
         :param pv_system_id (optional): Unique PV system ID,
             this can be provided or determined from a call to
             get_pvsystems_meta_data()
         :param httpx_client (optional)
         """
         self.access_key_id = access_key_id
         self.access_key_value = access_key_value
+        self.login_name = login_name
+        self.login_password = login_password
         self.pv_system_id = pv_system_id
         self.httpx_client = httpx_client or AsyncClient()
-
-    @property
-    def _common_headers(self):
-        return {
-            "Accept": "application/json",
+        self.jwt_data: dict = None
+        self._jwt_base_header = {
+            "Content-Type": "application/json-patch+json",
             "AccessKeyId": self.access_key_id,
             "AccessKeyValue": self.access_key_value,
+            "Accept": "application/json",
+            # "Accept-Language": "de-de",
+            "User-Agent": "Solar.web/921 CFNetwork/1410.0.3 Darwin/22.6.0",
         }
 
+    @property
+    def _common_headers(self):
+        if self._jwt_headers.get("Authorization"):
+            return self._jwt_headers
+        else:
+            return {
+                "Accept": "application/json",
+                "AccessKeyId": self.access_key_id,
+                "AccessKeyValue": self.access_key_value,
+            }
+
+    @property
+    def _jwt_headers(self):
+        return self._jwt_base_header
+
+    @_jwt_headers.setter
+    def _jwt_headers(self, new: dict):
+        return self._jwt_base_header.update(new)
+
+    def _jwt_del_header(self, key: str):
+        self._jwt_base_header.pop(key, None)
+
     async def _check_api_response(self, response):
         if response.status_code == 401:
-            _LOGGER.debug("Access unauthorised check solar.web access key values")
+            _LOGGER.warning(
+                "Access unauthorised check solar.web access key values or login password"
+            )
             raise NotAuthorizedException()
         if response.status_code == 404:
-            _LOGGER.debug("Item not found check your PV system ID")
+            _LOGGER.warning("Item not found check your PV system ID")
             raise NotFoundException()
 
         response.raise_for_status()
         # returns dict type not string
         return response.json()
 
+    async def login(self):
+        self._jwt_del_header("Authorization")
+        _LOGGER.debug("Obtaining JSON web token")
+        r = await self.httpx_client.post(
+            f"{SW_BASE_URL}/iam/jwt",
+            headers=self._jwt_headers,
+            json={
+                "userId": self.login_name,
+                "password": self.login_password,
+            },
+        )
+        self.jwt_data = await self._check_api_response(r)
+        _LOGGER.debug(f"JWT data returned: {self.jwt_data}")
+        self._jwt_headers = {"Authorization": "Bearer " + self.jwt_data.get("jwtToken")}
+
+    async def refresh_token(self, token: str = None):
+        refresh = self.jwt_data.get("refreshToken", token)
+        self._jwt_del_header("Authorization")
+        _LOGGER.debug(f"Obtaining JSON web token using refresh token: {refresh}")
+        r = await self.httpx_client.patch(
+            f"{SW_BASE_URL}/iam/jwt/{refresh}", headers={"accept": "application/json"}
+        )
+        self.jwt_data = await self._check_api_response(r)
+        _LOGGER.debug(f"JWT data returned: {self.jwt_data}")
+        self._jwt_headers = {"Authorization": "Bearer " + self.jwt_data.get("jwtToken")}
+
     @retry(
         wait=wait_random_exponential(multiplier=2, max=60),
         retry=retry_if_not_exception_type(
@@ -93,7 +150,7 @@ async def get_api_release_info(self) -> ReleaseInfo:
             model_data = ReleaseInfo.model_validate(json_data)
         except ValidationError as e:
             _LOGGER.error(
-                f"Unable to validate data receieved from SolarWeb api: '{json_data}'"
+                f"Unable to validate data receieved from SolarWeb api: {json_data}"
             )
             _LOGGER.error(e)
         return model_data
@@ -116,7 +173,7 @@ async def get_pvsystems_meta_data(self) -> list[PvSystemMetaData]:
             model_data = PvSystemsMetaData.model_validate(json_data).pvSystems
         except ValidationError as e:
             _LOGGER.error(
-                f"Unable to validate data receieved from SolarWeb api: '{json_data}'"
+                f"Unable to validate data receieved from SolarWeb api: {json_data}"
             )
             _LOGGER.error(e)
         return model_data
@@ -139,7 +196,7 @@ async def get_pvsystem_meta_data(self) -> PvSystemMetaData:
             model_data = PvSystemMetaData.model_validate(json_data)
         except ValidationError as e:
             _LOGGER.error(
-                f"Unable to validate data receieved from SolarWeb api: '{json_data}'"
+                f"Unable to validate data receieved from SolarWeb api: {json_data}"
             )
             _LOGGER.error(e)
         return model_data
@@ -162,7 +219,7 @@ async def get_devices_meta_data(self) -> list[DeviceMetaData]:
             model_data = DevicesMetaData.model_validate(json_data).devices
         except ValidationError as e:
             _LOGGER.error(
-                f"Unable to validate data receieved from SolarWeb api: '{json_data}'"
+                f"Unable to validate data receieved from SolarWeb api: {json_data}"
             )
             _LOGGER.error(e)
         return model_data
@@ -185,7 +242,7 @@ async def get_system_flow_data(self, tz: str = "zulu") -> PvSystemFlowData:
             model_data = PvSystemFlowData.model_validate(json_data)
         except ValidationError as e:
             _LOGGER.error(
-                f"Unable to validate data receieved from SolarWeb api: '{json_data}'"
+                f"Unable to validate data receieved from SolarWeb api: {json_data}"
             )
             _LOGGER.error(e)
         return model_data
@@ -213,7 +270,7 @@ async def get_system_aggr_data_v2(
             model_data = PvSystemAggrDataV2.model_validate(json_data)
         except ValidationError as e:
             _LOGGER.error(
-                f"Unable to validate data receieved from SolarWeb api: '{json_data}'"
+                f"Unable to validate data receieved from SolarWeb api: {json_data}"
             )
             _LOGGER.error(e)
         return model_data
@@ -241,7 +298,7 @@ async def get_hist_data(
             model_data = HistoricalValues.model_validate(json_data)
         except ValidationError as e:
             _LOGGER.error(
-                f"Unable to validate data receieved from SolarWeb api: '{json_data}'"
+                f"Unable to validate data receieved from SolarWeb api: {json_data}"
             )
             _LOGGER.error(e)
         return model_data

fronius_solarweb/schema/device.py

@@ -43,7 +43,7 @@ class DeviceMetaData(BaseModel):
     serialnumber: Optional[str] = None
     deviceTypeDetails: Optional[str] = None
     dataloggerId: Optional[str] = None
-    nodeType: Optional[str] = None
+    nodeType: Optional[int] = None
     numberMPPTrackers: Optional[int] = None
     numberPhases: Optional[int] = None
     peakPower: Optional[Power] = None

fronius_solarweb/schema/error.py

@@ -1,5 +1,3 @@
-from datetime import datetime
-
 from pydantic import BaseModel