New libbi action chargefromgrid

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
+[email protected]
+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/VERSION

@@ -1 +1 @@
-0.0.29
+0.0.30

pymyenergi/cli.py

@@ -27,8 +27,15 @@
 
 async def main(args):
     username = args.username or input("Please enter your hub serial number: ")
-    password = args.password or getpass()
-    conn = Connection(username, password)
+    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: ")
+    else:
+        app_password = ''
+    conn = Connection(username, password, app_password, app_email)
+    if app_email and app_password:
+        await conn.discoverLocations()
     if args.debug:
         logging.root.setLevel(logging.DEBUG)
     client = MyenergiClient(conn)
@@ -92,6 +99,19 @@ async def main(args):
                         sys.exit(f"A mode must be specifed, one of {modes}")
                     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("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")
@@ -148,7 +168,7 @@ async def main(args):
 
 def cli():
     config = configparser.ConfigParser()
-    config["hub"] = {"serial": "", "password": ""}
+    config["hub"] = {"serial": "", "password": "", "app_password": "", "app_email": ""}
     config.read([".myenergi.cfg", os.path.expanduser("~/.myenergi.cfg")])
     parser = argparse.ArgumentParser(prog="myenergi", description="myenergi CLI.")
     parser.add_argument(
@@ -163,6 +183,18 @@ def cli():
         dest="password",
         default=config.get("hub", "password").strip('"'),
     )
+    parser.add_argument(
+        "-a",
+        "--app_password",
+        dest="app_password",
+        default=config.get("hub", "app_password").strip('"'),
+    )
+    parser.add_argument(
+        "-e",
+        "--app_email",
+        dest="app_email",
+        default=config.get("hub", "app_email").strip('"'),
+    )
     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)
@@ -210,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"])
+    subparser_libbi.add_argument(
+        "action",
+        choices=[
+            "show",
+            "mode",
+            "priority",
+            "energy",
+            "chargefromgrid",
+            "chargetarget",
+        ],
+    )
     subparser_libbi.add_argument("arg", nargs="*")
 
     args = parser.parse_args()

pymyenergi/client.py

@@ -199,7 +199,6 @@ def power_charging(self):
     def power_battery(self):
         """Battery total power"""
         return self._totals.get(CT_BATTERY, 0)
-
 
     def find_device_name(self, key, default_value):
         """Find device or site name"""
@@ -248,6 +247,11 @@ async def refresh(self):
                         f"Updating {existing_device.kind} {existing_device.name}"
                     )
                     existing_device.data = device_data
+
+                # Update the extra information available on libbi
+                # this is the bit that requires OAuth
+                if existing_device.kind == LIBBI:
+                    await existing_device.refresh_extra()
         self._calculate_totals()
 
     async def refresh_history_today(self):

pymyenergi/connection.py

@@ -9,28 +9,39 @@
 
 import httpx
 
+from pycognito import Cognito
+
 from .exceptions import MyenergiException
 from .exceptions import TimeoutException
 from .exceptions import WrongCredentials
 
 _LOGGER = logging.getLogger(__name__)
-
+_USER_POOL_ID = 'eu-west-2_E57cCJB20'
+_CLIENT_ID = '2fup0dhufn5vurmprjkj599041'
 
 class Connection:
     """Connection to myenergi API."""
 
     def __init__(
-        self, username: Text = None, password: Text = None, timeout: int = 20
+        self, username: Text = None, password: Text = None, app_password: Text = None, app_email: Text = None, timeout: int = 20
     ) -> None:
         """Initialize connection object."""
         self.timeout = timeout
         self.director_url = "https://director.myenergi.net"
         self.base_url = None
+        self.oauth_base_url = "https://myaccount.myenergi.com"
         self.username = username
         self.password = password
+        self.app_password = app_password
+        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:
+            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")
 
     def _checkMyenergiServerURL(self, responseHeader):
@@ -45,50 +56,93 @@ def _checkMyenergiServerURL(self, responseHeader):
             )
             raise WrongCredentials()
 
-    async def send(self, method, url, json=None):
-        # If base URL has not been set, make a request to director to fetch it
+    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"]
+
+    def checkAndUpdateToken(self):
+        # 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 with httpx.AsyncClient(
-            auth=self.auth, headers=self.headers, timeout=self.timeout
-        ) as httpclient:
-            if self.base_url is None or self.do_query_asn:
-                _LOGGER.debug("Get Myenergi base url from director")
+    async def send(self, method, url, json=None, oauth=False):
+        # Use OAuth for myaccount.myenergi.com
+        if oauth:
+            # 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:
+                        _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:
+            # If base URL has not been set, make a request to director to fetch it
+            async with httpx.AsyncClient(
+                auth=self.auth, headers=self.headers, timeout=self.timeout
+            ) as httpclient:
+                if self.base_url is None or self.do_query_asn:
+                    _LOGGER.debug("Get Myenergi base url from director")
+                    try:
+                        directorUrl = self.director_url + "/cgi-jstatus-E"
+                        response = await httpclient.get(directorUrl)
+                    except Exception:
+                        _LOGGER.error("Myenergi server request problem")
+                        _LOGGER.debug(sys.exc_info()[0])
+                    else:
+                        self.do_query_asn = False
+                        self._checkMyenergiServerURL(response.headers)
+                theUrl = self.base_url + url
                 try:
-                    directorUrl = self.director_url + "/cgi-jstatus-E"
-                    response = await httpclient.get(directorUrl)
-                except Exception:
-                    _LOGGER.error("Myenergi server request problem")
-                    _LOGGER.debug(sys.exc_info()[0])
+                    _LOGGER.debug(f"{method} {url} {theUrl}")
+                    response = await httpclient.request(method, theUrl, json=json)
+                except httpx.ReadTimeout:
+                    # Make sure to query for ASN next request, might be a server problem
+                    self.do_query_asn = True
+                    raise TimeoutException()
                 else:
-                    self.do_query_asn = False
+                    _LOGGER.debug(f"GET status {response.status_code}")
                     self._checkMyenergiServerURL(response.headers)
-            theUrl = self.base_url + url
-            try:
-                _LOGGER.debug(f"{method} {url} {theUrl}")
-                response = await httpclient.request(method, theUrl, json=json)
-            except httpx.ReadTimeout:
-                # Make sure to query for ASN next request, might be a server problem
-                self.do_query_asn = True
-                raise TimeoutException()
-            else:
-                _LOGGER.debug(f"GET status {response.status_code}")
-                self._checkMyenergiServerURL(response.headers)
-                if response.status_code == 200:
-                    return response.json()
-                elif response.status_code == 401:
-                    raise WrongCredentials()
-                # Make sure to query for ASN next request, might be a server problem
-                self.do_query_asn = True
-                raise MyenergiException(response.status_code)
-
-    async def get(self, url):
-        return await self.send("GET", url)
-
-    async def post(self, url, data=None):
-        return await self.send("POST", url, data)
-
-    async def put(self, url, data=None):
-        return await self.send("PUT", url, data)
-
-    async def delete(self, url, data=None):
-        return await self.send("DELETE", url, data)
+                    if response.status_code == 200:
+                        return response.json()
+                    elif response.status_code == 401:
+                        raise WrongCredentials()
+                    # Make sure to query for ASN next request, might be a server problem
+                    self.do_query_asn = True
+                    raise MyenergiException(response.status_code)
+
+    async def get(self, url, data=None, oauth=False):
+        return await self.send("GET", url, data, oauth)
+
+    async def post(self, url, data=None, oauth=False):
+        return await self.send("POST", url, data, oauth)
+
+    async def put(self, url, data=None, oauth=False):
+        return await self.send("PUT", url, data, oauth)
+
+    async def delete(self, url, data=None, oauth=False):
+        return await self.send("DELETE", url, data, oauth)