Merge pull request #3 from trizmark/main

New libbi mode - 'export'

README.md

@@ -26,7 +26,7 @@ Setup will add a cli under the name myenergicli, see below for usage
 
 A simple cli is provided with this library.
 
-If no username or password is supplied as input arguments and no configuration file is found you will be prompted.
+If no username, password, app_email or app_password is supplied as input arguments and no configuration file is found you will be prompted.
 Conifguration file will be searched for in ./.myenergi.cfg and ~/.myenergi.cfg
 
 ### Example configuration file
@@ -35,18 +35,20 @@ Conifguration file will be searched for in ./.myenergi.cfg and ~/.myenergi.cfg
 [hub]
 serial=12345678
 password=yourpassword
+app_email=myemail@email.com
+app_password=yourapppassword
 ```
 
 ### CLI usage
 
 ```
-usage: myenergi [-h] [-u USERNAME] [-p PASSWORD] [-d] [-j]
-                {list,overview,zappi,eddi,harvi} ...
+usage: myenergi [-h] [-u USERNAME] [-p PASSWORD] [-e APP_EMAIL] [-a APP_PASSWORD] [-d] [-j]
+                {list,overview,zappi,eddi,harvi,libbi} ...
 
 myenergi CLI.
 
 positional arguments:
-  {list,overview,zappi,eddi,harvi}
+  {list,overview,zappi,eddi,harvi,libbi}
                         sub-command help
     list                list devices
     overview            show overview
@@ -59,6 +61,8 @@ optional arguments:
   -h, --help            show this help message and exit
   -u USERNAME, --username USERNAME
   -p PASSWORD, --password PASSWORD
+  -e APP_EMAIL, --app_email APP_EMAIL
+  -a APP_PASSWORD, --app_password APP_PASSWORD
   -d, --debug
   -j, --json
 ```
@@ -146,20 +150,23 @@ loop.run_until_complete(get_data())
 ```
 
 ## Libbi support
-Very early and basic support of Libbi.
+Currently supported features:
 
 - Reads a few values such as State of Charge, DCPV CT
 - Battery in and out energy
-- Gets and sets the current status
+- Gets and sets the current operating mode (normal/stopped/export)
 - Change priority of Libbi
+- Enable/Disable charging from the grid
+- Set charge target (in Wh)
 
 cli examples:
 ```
 myenergi libbi show
 myenergi libbi mode normal
-myenergi libbi mode stop
 myenergi libbi priority 1
 myenergi libbi energy
+myenergi libbi chargefromgrid false
+myenergi libbi chargetarget 10200
 ```
 
 

pymyenergi/cli.py

@@ -28,10 +28,14 @@
 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: ")
-    app_password = args.app_password or getpass(prompt="App password: ")
+    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 = ''
     conn = Connection(username, password, app_password, app_email)
-    await conn.discoverLocations()
+    if app_email and app_password:
+        await conn.discoverLocations()
     if args.debug:
         logging.root.setLevel(logging.DEBUG)
     client = MyenergiClient(conn)
@@ -96,10 +100,18 @@ async def main(args):
                     await device.set_operating_mode(args.arg[0])
                     print(f"Operating mode was set to {args.arg[0].capitalize()}")
                 elif args.action == "chargefromgrid" and args.command == LIBBI:
-                    if len(args.arg) < 1 or args.arg[0].capitalize() not in ["True", "False"]:
-                        sys.exit(f"A mode must be specifed, one of true or false")
+                    if len(args.arg) < 1 or args.arg[0].capitalize() not in [
+                        "True",
+                        "False",
+                    ]:
+                        sys.exit("A mode must be specifed, one of true or false")
                     await device.set_charge_from_grid(args.arg[0])
                     print(f"Charge from grid was set to {args.arg[0].capitalize()}")
+                elif args.action == "chargetarget" and args.command == LIBBI:
+                    if len(args.arg) < 1 or not args.arg[0].isnumeric():
+                        sys.exit("The charge target must be specified in Wh")
+                    await device.set_charge_target(args.arg[0])
+                    print(f"Charge target was set to {args.arg[0]}Wh")
                 elif args.action == "mingreen" and args.command == ZAPPI:
                     if len(args.arg) < 1:
                         sys.exit("A minimum green level must be provided")
@@ -230,7 +242,17 @@ def cli():
         LIBBI, help="use libbi --help for available commands"
     )
     subparser_libbi.add_argument("-s", "--serial", dest="serial", default=None)
-    subparser_libbi.add_argument("action", choices=["show","mode","priority","energy","chargefromgrid"])
+    subparser_libbi.add_argument(
+        "action",
+        choices=[
+            "show",
+            "mode",
+            "priority",
+            "energy",
+            "chargefromgrid",
+            "chargetarget",
+        ],
+    )
     subparser_libbi.add_argument("arg", nargs="*")
 
     args = parser.parse_args()

pymyenergi/connection.py

@@ -36,9 +36,10 @@ def __init__(
         self.app_email = app_email
         self.auth = httpx.DigestAuth(self.username, self.password)
         self.headers = {"User-Agent": "Wget/1.14 (linux-gnu)"}
-        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}"}
+        if self.app_email and 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}"}
         self.do_query_asn = True
         self.invitation_id = ''
         _LOGGER.debug("New connection created")
@@ -62,35 +63,41 @@ async def discoverLocations(self):
             self.invitation_id = locs["content"][0]["invitationData"]["invitationId"]
 
     def checkAndUpdateToken(self):
-        # check if we have to renew out token
-        self.oauth.check_token()
-        self.oauth_headers = {"Authorization": f"Bearer {self.oauth.access_token}"}
+        # check if we have oauth credentials
+        if self.app_email and self.app_password:
+            # check if we have to renew out token
+            self.oauth.check_token()
+            self.oauth_headers = {"Authorization": f"Bearer {self.oauth.access_token}"}
 
     async def send(self, method, url, json=None, oauth=False):
         # Use OAuth for myaccount.myenergi.com
         if oauth:
-            async with httpx.AsyncClient(
-                headers=self.oauth_headers, timeout=self.timeout
-            ) as httpclient:
-                theUrl = self.oauth_base_url + url
-                # if we have an invitiation id, we need to add that to the query
-                if (self.invitation_id != ""):
-                    if ("?" in theUrl):
-                        theUrl = theUrl + "&invitationId=" + self.invitation_id
+            # check if we have oauth credentials
+            if self.app_email and self.app_password:
+                async with httpx.AsyncClient(
+                    headers=self.oauth_headers, timeout=self.timeout
+                ) as httpclient:
+                    theUrl = self.oauth_base_url + url
+                    # if we have an invitiation id, we need to add that to the query
+                    if (self.invitation_id != ""):
+                        if ("?" in theUrl):
+                            theUrl = theUrl + "&invitationId=" + self.invitation_id
+                        else:
+                            theUrl = theUrl + "?invitationId=" + self.invitation_id
+                    try:
+                        _LOGGER.debug(f"{method} {url} {theUrl}")
+                        response = await httpclient.request(method, theUrl, json=json)
+                    except httpx.ReadTimeout:
+                        raise TimeoutException()
                     else:
-                        theUrl = theUrl + "?invitationId=" + self.invitation_id
-                try:
-                    _LOGGER.debug(f"{method} {url} {theUrl}")
-                    response = await httpclient.request(method, theUrl, json=json)
-                except httpx.ReadTimeout:
-                    raise TimeoutException()
-                else:
-                    _LOGGER.debug(f"{method} status {response.status_code}")
-                    if response.status_code == 200:
-                        return response.json()
-                    elif response.status_code == 401:
-                        raise WrongCredentials()
-                    raise MyenergiException(response.status_code)
+                        _LOGGER.debug(f"{method} status {response.status_code}")
+                        if response.status_code == 200:
+                            return response.json()
+                        elif response.status_code == 401:
+                            raise WrongCredentials()
+                        raise MyenergiException(response.status_code)
+            else:
+                _LOGGER.error("Trying to use OAuth without app credentials")
 
         # Use Digest Auth for director.myenergi.net and s18.myenergi.net
         else:

pymyenergi/libbi.py

@@ -7,27 +7,39 @@
 
 _LOGGER = logging.getLogger(__name__)
 
-MODE_NORMAL = 1
-MODE_STOPPED = 0
-
-STATES = { 0:'Off',
-           1:'On',
-           2:'Battery Full',
-           4:'Idle',
-           5:'Charging',
-           6:'Discharging',
-           7:'Duration Charging',
-           101:'Idle?',
-           102:'102',
-           104:'Battery Full?',
-           151:'FW Upgrade (ARM)',
-           156:'FW Upgrade (DSP)',
-           234:'Calibration Charge',
-           251:'FW Upgrade (DSP)',
-           252:'FW Upgrade (ARM)' }
-
-LIBBI_MODES = ["Stopped","Normal"]
-LIBBI_MODE_NAMES = ["STOP", "BALANCE"]
+STATES = {
+    0: "Off",
+    1: "On",
+    2: "Battery Full",
+    4: "Idle",
+    5: "Charging",
+    6: "Discharging",
+    7: "Duration Charging",
+    8: "Duration Drain",
+    12: "Target Charge",
+    51: "Boosting",
+    53: "Boosting",
+    55: "Boosting",
+    11: "Stopped",
+    101: "Battery Empty",
+    102: "Full",
+    104: "Full",
+    151: "FW Upgrade (ARM)",
+    156: "FW Upgrade (DSP)",
+    172: "BMS Charge Temperature Low",
+    234: "Calibration Charge",
+    251: "FW Upgrade (DSP)",
+    252: "FW Upgrade (ARM)",
+}
+
+LIBBI_MODES = ["Stopped", "Normal", "Export"]
+LIBBI_MODE_CONFIG = {
+    "Stopped": {"mode_int": 0, "mode_name": "STOP"},
+    "Normal": {"mode_int": 1, "mode_name": "BALANCE"},
+    "Export": {"mode_int": 5, "mode_name": "DRAIN"},
+}
+"""The myenergi app defines other modes as well (capture, charge, match), but these cannot be set"""
+
 
 class Libbi(BaseDevice):
     """Libbi Client for myenergi API."""
@@ -38,8 +50,20 @@ def __init__(self, connection: Connection, serialno, data={}) -> None:
         super().__init__(connection, serialno, data)
 
     async def refresh_extra(self):
-        chargeFromGrid = await self._connection.get("/api/AccountAccess/LibbiMode?serialNo=" + str(self.serial_number), oauth=True)
-        self._extra_data["charge_from_grid"] = chargeFromGrid["content"][str(self.serial_number)]
+        # only refresh this data if we have app credentials
+        if self._connection.app_email and self._connection.app_password:
+            chargeFromGrid = await self._connection.get(
+                "/api/AccountAccess/LibbiMode?serialNo=" + str(self.serial_number),
+                oauth=True,
+            )
+            self._extra_data["charge_from_grid"] = chargeFromGrid["content"][
+                str(self.serial_number)
+            ]
+            chargeTarget = await self._connection.get(
+                "/api/AccountAccess/" + str(self.serial_number) + "/LibbiChargeSetup",
+                oauth=True,
+            )
+            self._extra_data["charge_target"] = chargeTarget["content"]["energyTarget"]
 
     @property
     def kind(self):
@@ -142,12 +166,12 @@ def priority(self):
     @property
     def battery_size(self):
         """Battery size in kwh"""
-        return self._data.get("mbc", 0) /1000
+        return self._data.get("mbc", 0) / 1000
 
     @property
     def inverter_size(self):
         """Inverter size in kwh"""
-        return self._data.get("mic", 0) /1000
+        return self._data.get("mic", 0) / 1000
 
     @property
     def grid_import(self):
@@ -173,32 +197,43 @@ def battery_discharge(self):
     def generated(self):
         """Solar generation from history data"""
         return self.history_data.get("generated", 0)
-    
+
     @property
     def charge_from_grid(self):
         """Is charging from the grid enabled?"""
         return self._extra_data.get("charge_from_grid")
 
+    @property
+    def charge_target(self):
+        """Libbi charge target"""
+        return self._extra_data.get("charge_target", 0) / 1000
+
     @property
     def prefix(self):
         return "L"
 
+    def get_mode_description(self, mode: str):
+        """Get the mode name as returned by myenergi API. E.g. Normal mode is BALANCE"""
+        for k in LIBBI_MODE_CONFIG:
+            if LIBBI_MODE_CONFIG[k]["mode_name"] == mode:
+                return k
+        return "???"
 
     async def set_operating_mode(self, mode: str):
-        """Stopped or normal mode"""
-        print("current mode", self._data["lmo"])
-        mode_int = LIBBI_MODES.index(mode.capitalize())
+        """Set operating mode"""
+        print("current mode", self.get_mode_description(self._data["lmo"]))
+        mode_int = LIBBI_MODE_CONFIG[mode.capitalize()]["mode_int"]
         await self._connection.get(
             f"/cgi-libbi-mode-{self.prefix}{self._serialno}-{mode_int}"
         )
-        self._data["lmo"] = LIBBI_MODE_NAMES[mode_int]
+        self._data["lmo"] = LIBBI_MODE_CONFIG[mode.capitalize()]["mode_name"]
         return True
 
     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
+            oauth=True,
         )
         self._extra_data["charge_from_grid"] = charge_from_grid
         return True
@@ -211,6 +246,15 @@ async def set_priority(self, priority):
         self._data["pri"] = int(priority)
         return True
 
+    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
+
     def show(self, short_format=False):
         """Returns a string with all data in human readable format"""
         ret = ""
@@ -228,9 +272,14 @@ def show(self, short_format=False):
         ret = ret + f"State of Charge: {self.state_of_charge}%\n"
         ret = ret + f"Generating: {self.power_generated}W\n"
         ret = ret + f"Grid: {self.power_grid}W\n"
-        ret = ret + f"Status : {self.status}\n"
-        ret = ret + f"Local Mode : {self.local_mode}\n"
-        ret = ret + f"Charge from Grid: {self.charge_from_grid}\n"
+        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"
+        else:
+            ret = ret + "Disabled\n"
+        ret = ret + f"Charge target: {self.charge_target}kWh\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"
@@ -239,4 +288,6 @@ def show(self, short_format=False):
         ret = ret + f"CT 6 {self.ct6.name} {self.ct6.power}W phase {self.ct6.phase}\n"
         for key in self.ct_keys:
             ret = ret + f"Energy {key} {self.history_data.get(key, 0)}Wh\n"
+        if not self._connection.app_email or not self._connection.app_password:
+            ret += "No app credentials provided - the above information might not be totally accurate\n"
         return ret