Merge branch 'main' of https://github.com/Woyken/py-electrolux-ocp in…

…to main

.github/workflows/black.yml

@@ -0,0 +1,10 @@
+name: Lint
+
+on: [push, pull_request]
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: psf/black@stable

.github/workflows/publish-workflow.yml

@@ -2,6 +2,10 @@ name: Publish Python 🐍 distribution 📦 to PyPI and TestPyPI
 
 on: push
 
+concurrency:
+  cancel-in-progress: true
+  group: publish to PyPi
+
 jobs:
   build:
     name: Build distribution 📦

setup.py

@@ -24,6 +24,6 @@
     package_dir={"": "src"},
     packages=setuptools.find_packages(where="src"),
     url="https://github.com/Woyken/py-electrolux-ocp",
-    version="0.0.6",
+    version="0.0.7",
     zip_safe=False,
 )

src/pyelectroluxocp/apiClient.py

@@ -240,7 +240,7 @@ async def get_appliances_list(
             data: list[ApplienceStatusResponse] = response_json
             return data
 
-    async def get_appliance_status(
+    async def get_appliance_state(
         self, base_url: str, token: str, id: str, include_metadata: bool
     ):
         _LOGGER.debug(

src/pyelectroluxocp/getCapabilitiesTODO.py

@@ -258,9 +258,9 @@ async def main():
     capabilities = await client.get_appliance_capabilities(
         appliances[0].get("applianceId")
     )
-    status = await client.get_appliance_status(appliances[0].get("applianceId"), False)
-    # print("aaaaaaaaaaaaaaaaaaaaaaaaaaaa", json.dumps(status))
-    print(json.dumps(parse_capabilities(capabilities, status)))
+    state = await client.get_appliance_state(appliances[0].get("applianceId"), False)
+    # print("aaaaaaaaaaaaaaaaaaaaaaaaaaaa", json.dumps(state))
+    print(json.dumps(parse_capabilities(capabilities, state)))
 
 
 asyncio.run(main())

src/pyelectroluxocp/gigyaClient.py

@@ -199,6 +199,15 @@ async def login_session(self, username: str, password: str, gmid: str, ucid: str
             )
             response.raise_for_status()
             data: LoginResponse = response_json
+            try:
+                if data["errorCode"] > 0:
+                    error_code = data["errorCode"]
+                    error_details = data["errorDetails"]
+                    raise LoginError(
+                        f"Error during login: Code {error_code} ({error_details})"
+                    )
+            except KeyError:
+                raise LoginError("Unknown error during login")
             return data
 
     async def get_JWT(
@@ -269,3 +278,9 @@ async def __aexit__(
         exc_tb: Optional[TracebackType],
     ) -> Optional[bool]:
         await self.close()
+
+
+class LoginError(Exception):
+    """Raised when login fails"""
+
+    pass

src/pyelectroluxocp/gigyaModels.py

@@ -1,5 +1,6 @@
 from typing import TypedDict
 
+
 class SocializeGetIdsResponse(TypedDict):
     callId: str
     errorCode: int
@@ -18,11 +19,13 @@ class LoginResponseProfile(TypedDict):
     country: str
     email: str
 
+
 class LoginResponseSessionInfo(TypedDict):
     sessionToken: str
     sessionSecret: str
     expires_in: str
 
+
 class LoginResponse(TypedDict):
     callId: str
     errorCode: int

src/pyelectroluxocp/oneAppApi.py

@@ -1,5 +1,5 @@
 import asyncio
-from json import dumps
+from json import dump, dumps
 from types import TracebackType
 from typing import Any, Callable, Dict, Optional, Type
 from aiohttp import ClientSession
@@ -137,17 +137,17 @@ async def get_appliances_list(self, include_metadata: bool = False):
         )
         return result
 
-    async def get_appliance_status(self, id: str, include_metadata: bool = False):
-        """Get current status of appliance by id"""
+    async def get_appliance_state(self, id: str, include_metadata: bool = False):
+        """Get current state of appliance by id"""
         _LOGGER.debug(
-            "get_appliance_capabilities(), id: %s, include_metadata: %s",
+            "get_appliance_state(), id: %s, include_metadata: %s",
             id,
             include_metadata,
         )
         token = await self._get_formatted_user_token()
         base_url = await self._get_base_url()
 
-        result = await self._api_client.get_appliance_status(
+        result = await self._api_client.get_appliance_state(
             base_url, token, id, include_metadata
         )
         return result
@@ -237,6 +237,67 @@ async def async_impl():
 
         await self.connect_websocket(appliance_ids)
 
+    async def debug_dump_everything_to_files(
+        self, export_location: Optional[str] = None
+    ):
+        """DEBUG, Fetch everything and dump to temporary files. Default export location: ~/py-electrolux-ocp-dump/..."""
+        from pathlib import Path
+
+        home = Path.home() if export_location is None else Path(export_location)
+        dump_path = home.joinpath("py-electrolux-ocp-dump")
+        from shutil import rmtree
+
+        rmtree(dump_path, ignore_errors=True)
+        dump_path.mkdir(exist_ok=True)
+
+        user_metadata = await self.get_user_metadata()
+        with open(dump_path.joinpath("userMetadata.json"), "w", encoding="utf-8") as f:
+            dump(user_metadata, f, ensure_ascii=False, indent=4)
+
+        appliances_list = await self.get_appliances_list(True)
+        with open(
+            dump_path.joinpath("appliancesList.json"), "w", encoding="utf-8"
+        ) as f:
+            dump(appliances_list, f, ensure_ascii=False, indent=4)
+
+        appliances_info = await self.get_appliances_info(
+            [x["applianceId"] for x in appliances_list]
+        )
+        with open(
+            dump_path.joinpath("appliancesInfo.json"), "w", encoding="utf-8"
+        ) as f:
+            dump(appliances_info, f, ensure_ascii=False, indent=4)
+
+        for appliance in appliances_list:
+            import base64
+
+            appliance_id_folder_name = base64.urlsafe_b64encode(
+                appliance["applianceId"].encode()
+            ).decode()
+            dump_path_appliance = dump_path.joinpath(appliance_id_folder_name)
+            dump_path_appliance.mkdir(exist_ok=True)
+            appliance_capabilities = await self.get_appliance_capabilities(
+                appliance["applianceId"]
+            )
+            with open(
+                dump_path_appliance.joinpath("applianceCapabilities.json"),
+                "w",
+                encoding="utf-8",
+            ) as f:
+                dump(appliance_capabilities, f, ensure_ascii=False, indent=4)
+
+            appliance_state = await self.get_appliance_state(
+                appliance["applianceId"], True
+            )
+            with open(
+                dump_path_appliance.joinpath("applianceState.json"),
+                "w",
+                encoding="utf-8",
+            ) as f:
+                dump(appliance_state, f, ensure_ascii=False, indent=4)
+
+        print("DUMP generated at", dump_path)
+
     async def close(self) -> None:
         """Dispose session and dependencies"""
         if self._gigya_client: