Added enphase_energy.py unit test

envoy_logger/cli.py

@@ -1,7 +1,7 @@
 import logging
 import argparse
 
-from . import enphaseenergy
+from .enphase_energy import EnphaseEnergy
 from .sampling_loop import SamplingLoop
 from .config import load_config
 
@@ -17,10 +17,12 @@ def main() -> None:
 
     config = load_config(args.config_path)
 
-    envoy_token = enphaseenergy.get_token(
-        config.enphase_email, config.enphase_password, config.envoy_serial
+    enphase_energy = EnphaseEnergy(
+        email=config.enphase_email,
+        password=config.enphase_password,
+        envoy_serial=config.envoy_serial,
     )
 
-    sampling_loop = SamplingLoop(envoy_token, config)
+    sampling_loop = SamplingLoop(enphase_energy, config)
 
     sampling_loop.run()

envoy_logger/config.py

@@ -1,5 +1,6 @@
 import logging
 import sys
+from typing import Dict, Optional
 
 import yaml
 from influxdb_client import Point
@@ -10,26 +11,24 @@
 class Config:
     def __init__(self, data) -> None:
         try:
-            self.enphase_email = data["enphaseenergy"]["email"]  # type: str
-            self.enphase_password = data["enphaseenergy"]["password"]  # type: str
+            self.enphase_email: str = data["enphaseenergy"]["email"]
+            self.enphase_password: str = data["enphaseenergy"]["password"]
 
             self.envoy_serial = str(data["envoy"]["serial"])
-            self.envoy_url = data["envoy"].get(
-                "url", "https://envoy.local"
-            )  # type: str
-            self.source_tag = data["envoy"].get("tag", "envoy")  # type: str
-
-            self.influxdb_url = data["influxdb"]["url"]  # type: str
-            self.influxdb_token = data["influxdb"]["token"]  # type: str
-            self.influxdb_org = data["influxdb"].get("org", "home")  # type: str
-
-            bucket = data["influxdb"].get("bucket", None)
-            bucket_lr = data["influxdb"].get("bucket_lr", None)
-            bucket_hr = data["influxdb"].get("bucket_hr", None)
-            self.influxdb_bucket_lr = bucket_lr or bucket
-            self.influxdb_bucket_hr = bucket_hr or bucket
-
-            self.inverters = {}  # type: Dict[str, InverterConfig]
+            self.envoy_url: str = data["envoy"].get("url", "https://envoy.local")
+            self.source_tag: str = data["envoy"].get("tag", "envoy")
+
+            self.influxdb_url: str = data["influxdb"]["url"]
+            self.influxdb_token: str = data["influxdb"]["token"]
+            self.influxdb_org: str = data["influxdb"].get("org", "home")
+
+            bucket: Optional[str] = data["influxdb"].get("bucket", None)
+            bucket_lr: Optional[str] = data["influxdb"].get("bucket_lr", None)
+            bucket_hr: Optional[str] = data["influxdb"].get("bucket_hr", None)
+            self.influxdb_bucket_lr: str = bucket_lr or bucket
+            self.influxdb_bucket_hr: str = bucket_hr or bucket
+
+            self.inverters: Dict[str, InverterConfig] = {}
             for serial, inverter_data in data.get("inverters", {}).items():
                 serial = str(serial)
                 self.inverters[serial] = InverterConfig(inverter_data, serial)

envoy_logger/enphase_energy.py

@@ -0,0 +1,125 @@
+from dataclasses import dataclass
+from typing import Optional
+from datetime import datetime, timedelta
+import json
+import base64
+import os
+import logging
+
+import requests
+from appdirs import user_cache_dir
+
+LOG = logging.getLogger("enphaseenergy")
+
+
+@dataclass(frozen=True)
+class EnphaseEnergy:
+    email: str
+    password: str
+    envoy_serial: str
+
+    def get_token(self) -> str:
+        """
+        Do whatever it takes to get a token
+        """
+        token = self._get_cached_token()
+        if token is None:
+            # cached token does not exist. Get a new one
+            token = self._get_new_token()
+            self._save_token_to_cache(token)
+
+        exp = self._token_expiration_date(token)
+        time_left = exp - datetime.now()
+        if time_left < timedelta(days=1):
+            # token will expire soon. get a new one
+            LOG.info("Token will expire soon. Getting a new one")
+            token = self._get_new_token()
+            self._save_token_to_cache(token)
+
+        return token
+
+    def _get_cached_token(self) -> Optional[str]:
+        path = self._get_token_cache_path()
+        if not os.path.exists(path):
+            return None
+        with open(path, "r", encoding="utf-8") as f:
+            LOG.info("Using cached token from: %s", path)
+            return f.read()
+
+    def _get_token_cache_path(self) -> str:
+        return os.path.join(
+            user_cache_dir("enphase-envoy"), f"{self.envoy_serial}.token"
+        )
+
+    def _get_new_token(self) -> str:
+        """
+        Login to enphaseenergy.com and return an access token for the envoy.
+        """
+        session_id = self._login_enphaseenergy()
+
+        LOG.info("Downloading new access token for envoy S/N: %s", self.envoy_serial)
+        # Get the token
+        json_data = {
+            "session_id": session_id,
+            "serial_num": self.envoy_serial,
+            "username": self.email,
+        }
+        response = requests.post(
+            "https://entrez.enphaseenergy.com/tokens",
+            json=json_data,
+            timeout=30,
+        )
+        response.raise_for_status()
+        return response.text
+
+    def _login_enphaseenergy(self) -> str:
+        LOG.info("Logging into enphaseenergy.com as %s", self.email)
+        # Login and get session ID
+        files = {
+            "user[email]": (None, self.email),
+            "user[password]": (None, self.password),
+        }
+        url = "https://enlighten.enphaseenergy.com/login/login.json?"
+        response = requests.post(
+            url,
+            files=files,
+            timeout=30,
+        )
+        response.raise_for_status()
+        resp = response.json()
+        return resp["session_id"]
+
+    def _save_token_to_cache(self, token: str) -> None:
+        path = self._get_token_cache_path()
+        LOG.info("Caching token to: %s", path)
+        parent_dir = os.path.dirname(path)
+        if not os.path.exists(parent_dir):
+            os.mkdir(parent_dir)
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(token)
+
+    def _token_expiration_date(self, token: str) -> datetime:
+        jwt = {}
+
+        # The token is a string containing two base64 encoded segments delimited by '.', the third token
+        # is not considered here.
+        # The first two segments parse to JSON and contain information about the logged in user.
+        #
+        # Example response here:
+        # {'kid': 'tokenid', 'typ': 'JWT', 'alg': 'ES256', 'aud': 'account#', 'iss': 'Entrez',
+        #   'enphaseUser': 'owner', 'exp': 1732674920, 'iat': 1701138920, 'jti': 'something',
+        #   'username': '[email protected]'}
+        #
+        # We are specifically looking to extract 'exp'
+        for token_segment in token.split(".")[0:2]:
+            # Append equals to the end of the segment so it can be decoded properly by b64decode
+            res = len(token_segment) % 4
+            if res != 0:
+                token_segment += "=" * (4 - res)
+
+            segment_json = base64.b64decode(token_segment)
+            jwt.update(json.loads(segment_json))
+
+        exp = datetime.fromtimestamp(jwt["exp"])
+
+        return exp

envoy_logger/sampling_loop.py

@@ -8,19 +8,26 @@
 from influxdb_client import WritePrecision, InfluxDBClient, Point
 from influxdb_client.client.write_api import SYNCHRONOUS
 
+from .enphase_energy import EnphaseEnergy
+
 from . import envoy
 
 
-from .model import SampleData, PowerSample, InverterSample, filter_new_inverter_data
+from .model import (
+    SampleData,
+    PowerSample,
+    InverterSample,
+    filter_new_inverter_data,
+)
 from .config import Config
 
 
 class SamplingLoop:
     interval = 5
 
-    def __init__(self, token: str, config: Config) -> None:
+    def __init__(self, enphase_energy: EnphaseEnergy, config: Config) -> None:
         self.config = config
-        self.session_id = envoy.login(self.config.envoy_url, token)
+        self.session_id = envoy.login(self.config.envoy_url, enphase_energy.get_token())
 
         influxdb_client = InfluxDBClient(
             url=config.influxdb_url,

test.sh

@@ -2,6 +2,8 @@
 set -eou pipefail
 cd "$(dirname "${0}")"
 
+rm -rf ~/.cache/enphase-envoy/
+
 poetry run black --check envoy_logger tests
 poetry run flake8 envoy_logger tests
 poetry run coverage run -m pytest -s

tests/test_cli.py

@@ -5,7 +5,7 @@
 from envoy_logger import cli
 
 
-@mock.patch("envoy_logger.enphaseenergy.get_token")
+@mock.patch("envoy_logger.enphase_energy.EnphaseEnergy")
 @mock.patch("envoy_logger.cli.SamplingLoop")
 @mock.patch("sys.argv", ["--config", "./docs/config.yml"])
 class TestCli(unittest.TestCase):