OAuth improvements and new status codes

README.md

@@ -42,7 +42,7 @@ app_password=yourapppassword
 ### CLI usage
 
 ```
-usage: myenergi [-h] [-u USERNAME] [-p PASSWORD] [-e APP_EMAIL] [-a APP_PASSWORD] [-d] [-j]
+usage: myenergi [-h] [-u USERNAME] [-p PASSWORD] [-e APP_EMAIL] [-a APP_PASSWORD] [-d] [-j] [--skip-oauth]
                 {list,overview,zappi,eddi,harvi,libbi} ...
 
 myenergi CLI.

pymyenergi/cli.py

@@ -13,13 +13,13 @@
 from pymyenergi.eddi import BOOST_TARGETS
 from pymyenergi.eddi import EDDI_MODES
 from pymyenergi.exceptions import WrongCredentials
-from pymyenergi.zappi import CHARGE_MODES
 from pymyenergi.libbi import LIBBI_MODES
+from pymyenergi.zappi import CHARGE_MODES
 
 from . import EDDI
 from . import HARVI
-from . import ZAPPI
 from . import LIBBI
+from . import ZAPPI
 
 logging.basicConfig()
 logging.root.setLevel(logging.WARNING)
@@ -28,11 +28,15 @@
 async def main(args):
     username = args.username or input("Please enter your hub serial number: ")
     password = args.password or getpass(prompt="Password (apikey): ")
-    app_email = args.app_email or input("App email (enter to skip; only needed for libbi): ")
-    if app_email:
-        app_password = args.app_password or getpass(prompt="App password: ")
+    if not args.skip_oauth:
+        app_email = args.app_email or input("App email (enter to skip; only needed for libbi): ")
+        if app_email:
+            app_password = args.app_password or getpass(prompt="App password: ")
+        else:
+            app_password = ""
     else:
-        app_password = ''
+        app_email = ""
+        app_password = ""
     conn = Connection(username, password, app_password, app_email)
     if app_email and app_password:
         await conn.discoverLocations()
@@ -195,6 +199,7 @@ def cli():
         dest="app_email",
         default=config.get("hub", "app_email").strip('"'),
     )
+    parser.add_argument("--skip-oauth", dest="skip_oauth", action="store_true", default=False)
     parser.add_argument("-d", "--debug", dest="debug", action="store_true")
     parser.add_argument("-j", "--json", dest="json", action="store_true", default=False)
     parser.add_argument("--version", dest="version", action="store_true", default=False)

pymyenergi/client.py

@@ -12,8 +12,8 @@
 from . import EDDI
 from . import FREQUENCY_GRID
 from . import HARVI
-from . import LIBBI
 from . import HOUR
+from . import LIBBI
 from . import VOLTAGE_GRID
 from . import ZAPPI
 from .eddi import Eddi

pymyenergi/connection.py

@@ -8,7 +8,6 @@
 from typing import Text
 
 import httpx
-
 from pycognito import Cognito
 
 from .exceptions import MyenergiException
@@ -19,6 +18,7 @@
 _USER_POOL_ID = 'eu-west-2_E57cCJB20'
 _CLIENT_ID = '2fup0dhufn5vurmprjkj599041'
 
+
 class Connection:
     """Connection to myenergi API."""
 
@@ -36,7 +36,7 @@ def __init__(
         self.app_email = app_email
         self.auth = httpx.DigestAuth(self.username, self.password)
         self.headers = {"User-Agent": "Wget/1.14 (linux-gnu)"}
-        if self.app_email and app_password:
+        if self.app_email and self.app_password:
             self.oauth = Cognito(_USER_POOL_ID, _CLIENT_ID, username=self.app_email)
             self.oauth.authenticate(password=self.app_password)
             self.oauth_headers = {"Authorization": f"Bearer {self.oauth.access_token}"}
@@ -57,10 +57,11 @@ def _checkMyenergiServerURL(self, responseHeader):
             raise WrongCredentials()
 
     async def discoverLocations(self):
-        locs = await self.get("/api/Location", oauth=True)
-        # check if guest location - use the first location by default
-        if locs["content"][0]["isGuestLocation"] == True:
-            self.invitation_id = locs["content"][0]["invitationData"]["invitationId"]
+        if self.app_email and self.app_password:
+            locs = await self.get("/api/Location", oauth=True)
+            # check if guest location - use the first location by default
+            if locs["content"][0]["isGuestLocation"] == True:
+                self.invitation_id = locs["content"][0]["invitationData"]["invitationId"]
 
     def checkAndUpdateToken(self):
         # check if we have oauth credentials

pymyenergi/libbi.py

@@ -1,12 +1,8 @@
-import logging
-
 from pymyenergi.connection import Connection
 
 from . import LIBBI
 from .base_device import BaseDevice
 
-_LOGGER = logging.getLogger(__name__)
-
 STATES = {
     0: "Off",
     1: "On",
@@ -27,9 +23,11 @@
     151: "FW Upgrade (ARM)",
     156: "FW Upgrade (DSP)",
     172: "BMS Charge Temperature Low",
+    176: "BMS Updating",
     234: "Calibration Charge",
     251: "FW Upgrade (DSP)",
     252: "FW Upgrade (ARM)",
+    253: "BMS Upgrading",
 }
 
 LIBBI_MODES = ["Stopped", "Normal", "Export"]
@@ -83,10 +81,6 @@ def local_mode(self):
         """Get current known status"""
         return self._data.get("lmo", 1)
 
-    @property
-    def prefix(self):
-        return "E"
-
     @property
     def ct_keys(self):
         """Return CT key names that are not none"""
@@ -201,12 +195,18 @@ def generated(self):
     @property
     def charge_from_grid(self):
         """Is charging from the grid enabled?"""
-        return self._extra_data.get("charge_from_grid")
+        if self._connection.app_email and self._connection.app_password:
+            return self._extra_data.get("charge_from_grid")
+        else:
+            return None
 
     @property
     def charge_target(self):
         """Libbi charge target"""
-        return self._extra_data.get("charge_target", 0) / 1000
+        if self._connection.app_email and self._connection.app_password:
+            return self._extra_data.get("charge_target", 0) / 1000
+        else:
+            return None
 
     @property
     def prefix(self):
@@ -231,12 +231,15 @@ async def set_operating_mode(self, mode: str):
 
     async def set_charge_from_grid(self, charge_from_grid: bool):
         """Set charge from grid"""
-        await self._connection.put(
-            f"/api/AccountAccess/LibbiMode?chargeFromGrid={charge_from_grid}&serialNo={self._serialno}",
-            oauth=True,
-        )
-        self._extra_data["charge_from_grid"] = charge_from_grid
-        return True
+        if self._connection.app_email and self._connection.app_password:
+            await self._connection.put(
+                f"/api/AccountAccess/LibbiMode?chargeFromGrid={charge_from_grid}&serialNo={self._serialno}",
+                oauth=True,
+            )
+            self._extra_data["charge_from_grid"] = charge_from_grid
+            return True
+        else:
+            return False
 
     async def set_priority(self, priority):
         """Set device priority"""
@@ -248,12 +251,15 @@ async def set_priority(self, priority):
 
     async def set_charge_target(self, charge_target: float):
         """Set charge target"""
-        await self._connection.put(
-            f"/api/AccountAccess/{self._serialno}/TargetEnergy?targetEnergy={charge_target}",
-            oauth=True,
-        )
-        self._extra_data["charge_target"] = charge_target
-        return True
+        if self._connection.app_email and self._connection.app_password:
+            await self._connection.put(
+                f"/api/AccountAccess/{self._serialno}/TargetEnergy?targetEnergy={charge_target}",
+                oauth=True,
+            )
+            self._extra_data["charge_target"] = charge_target
+            return True
+        else:
+            return False
 
     def show(self, short_format=False):
         """Returns a string with all data in human readable format"""
@@ -275,11 +281,18 @@ def show(self, short_format=False):
         ret = ret + f"Status: {self.status}\n"
         ret = ret + "Local Mode: " + self.get_mode_description(self.local_mode) + "\n"
         ret = ret + "Charge from Grid: "
-        if self.charge_from_grid:
-            ret = ret + "Enabled\n"
+        if self.charge_from_grid is not None:
+            if self.charge_from_grid:
+                ret = ret + "Enabled\n"
+            else:
+                ret = ret + "Disabled\n"
+        else:
+            ret = ret + f"<unavailable>\n"
+        ret = ret + f"Charge target: "
+        if self.charge_target is not None:
+            ret = ret + f"{self.charge_target}kWh\n"
         else:
-            ret = ret + "Disabled\n"
-        ret = ret + f"Charge target: {self.charge_target}kWh\n"
+            ret = ret + f"<unavailable>\n"
         ret = ret + f"CT 1 {self.ct1.name} {self.ct1.power}W phase {self.ct1.phase}\n"
         ret = ret + f"CT 2 {self.ct2.name} {self.ct2.power}W phase {self.ct2.phase}\n"
         ret = ret + f"CT 3 {self.ct3.name} {self.ct3.power}W phase {self.ct3.phase}\n"

pymyenergi/zappi.py

@@ -16,6 +16,7 @@
 }
 SINGLE_PHASE = "SINGLE_PHASE"
 
+
 class Zappi(BaseDevice):
     """Zappi Client for myenergi API."""
 

tests/fixtures/libbi.json

@@ -38,4 +38,4 @@
   "fwv": "3702S5.041",
   "newAppAvailable": "False",
   "newBootloaderAvailable": "False"
-}
+}

tests/test_client.py

@@ -2,8 +2,8 @@
 from pymyenergi.client import MyenergiClient
 from pymyenergi.eddi import Eddi
 from pymyenergi.harvi import Harvi
-from pymyenergi.zappi import Zappi
 from pymyenergi.libbi import Libbi
+from pymyenergi.zappi import Zappi
 
 # All test coroutines will be treated as marked.
 pytestmark = pytest.mark.asyncio