Initial add

.gitignore

@@ -0,0 +1,5 @@
+.env
+.venv/
+.vscode
+credentials.json
+envoy-logger-config.yml

README.md

@@ -0,0 +1,4 @@
+Miscellaneous scripts used to gather data from an Enphase Envoy/Gateway
+
+This stuff is meant for research.  If you're looking for an organized project
+that queries data from an Envoy to save and graph, check out amykyta3/envoy-logger.

envoy.py

@@ -0,0 +1,201 @@
+#!/usr/bin/env python3
+
+""" envoy.py
+
+Read data from a local Enphase Envoy and write it to stdout (print it)
+
+Usage:
+
+    envoy.py <endpoint>
+
+...where <endpoint> is one of:
+
+    meter_details
+    meter_readings
+    inverter_production_data
+    meter_live_data
+    load_consumption_data
+    home_json
+    production_json
+    production_json_details
+    inventory_json
+    inventory_json_deleted
+
+Before running this script you must create a ".env" file in the same directory
+as this script with:
+
+    ENLIGHTEN_USERNAME="Enlighten username (email address)"
+    ENLIGHTEN_PASSWORD="Enlighten password"
+    ENVOY_SERIAL="serial number of your Envoy"
+    ENVOY_HOSTNAME="hostname or IP address of your Envoy/Gateway"
+
+Alternatively, you can use environment variables of the same names.
+
+!!! SECURITY WARNING !!!
+
+This script will create a "credentials.json" file in the same directory as this
+script that includes an API token that can be used to access your Envoy/Gateway.
+Additionally, the .env file has similar credentials.  Be careful!!  You may want
+to create an additional Enlighten account and grant it read-only access to your
+Envoy.
+
+"""
+
+import datetime
+import json
+import os
+import sys
+import traceback
+from typing import Any
+
+import jwt
+import requests
+from dotenv import dotenv_values
+from urllib3.exceptions import InsecureRequestWarning
+
+# Envoy uses self-signed SSL certificate
+requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
+
+DATA_URIS = {
+    "meter_details": "/ivp/meters",
+    "meter_readings": "/ivp/meters/readings",
+    "inverter_production_data": "/api/v1/production/inverters",
+    "meter_live_data": "/ivp/livedata/status",
+    "load_consumption_data": "/ivp/meters/reports/consumption",
+    "home_json": "/home.json",
+    "production_json": "/production.json",
+    "production_json_details": "/production.jsoni?details=1",
+    "inventory_json": "/inventory.json",
+    "inventory_json_deleted": "/inventory.json?deleted=1",
+}
+
+CREDENTIALS_FILE = os.path.join(
+    os.path.dirname(os.path.realpath(__file__)), "credentials.json"
+)
+
+
+def get_token(config: dict = dotenv_values()) -> dict:
+    s = requests.Session()
+
+    # Initiate a login session
+    # This is a form post, hence "data=" (not "json=")
+    # A JSON string is returned in the response body
+    r = s.post(
+        "https://enlighten.enphaseenergy.com/login/login.json?",
+        data={
+            "user[email]": config["ENLIGHTEN_USERNAME"],
+            "user[password]": config["ENLIGHTEN_PASSWORD"],
+        },
+    )
+
+    # Example Response JSON:
+    #
+    # {
+    #     "message": "success",
+    #     "session_id": "blahblahblahhexstring",
+    #     "manager_token": "big.old_long_jwt_token_here",
+    #     "is_consumer": True
+    # }
+    login_json = json.loads(r.text)
+
+    # Retrieve a token
+    r = s.post(
+        "https://entrez.enphaseenergy.com/tokens",
+        json={
+            "session_id": login_json["session_id"],
+            "serial_num": config["ENVOY_SERIAL"],
+            "username": config["ENLIGHTEN_USERNAME"],
+        },
+    )
+
+    token = r.text
+
+    # Extract the expiration
+    decoded = jwt.decode(
+        token,
+        options={"verify_signature": False},
+    )
+
+    expiration = datetime.datetime.fromtimestamp(decoded["exp"])
+
+    # Write credentials file
+    with open(CREDENTIALS_FILE, "w", encoding="utf-8") as f:
+        json.dump(
+            {
+                "token": token,
+                "expires": str(expiration),
+                "expires_epoch": decoded["exp"],
+            },
+            f,
+            ensure_ascii=False,
+            indent=4,
+        )
+
+    return {
+        "token": token,
+        "expires": str(expiration),
+        "expires_epoch": decoded["exp"],
+    }
+
+
+def read_token(config: dict = dotenv_values()) -> dict:
+    with open(CREDENTIALS_FILE, "r", encoding="utf-8") as f:
+        credentials = json.load(f)
+
+    expiration = datetime.datetime.fromtimestamp(credentials["expires_epoch"])
+
+    if expiration <= datetime.datetime.now():
+        # Token is expired, attempt refresh
+        credentials = get_token(config)
+
+    return credentials
+
+
+def get_data(envoy_hostname: str, endpoint: str, token: str) -> Any:
+    r = requests.get(
+        f"https://{envoy_hostname}{DATA_URIS[endpoint]}",
+        headers={
+            "Accept": "application/json",
+            "Authorization": f"Bearer {token}",
+        },
+        verify=False,
+    )
+    return r.json()
+
+
+def main(endpoint: str, config: dict = dotenv_values()) -> None:
+    if endpoint not in DATA_URIS.keys():
+        raise RuntimeError("Unknown endpoint")
+
+    if os.path.isfile(CREDENTIALS_FILE):
+        credentials = read_token(config)
+    else:
+        credentials = get_token(config)
+
+    print(
+        json.dumps(
+            get_data(
+                envoy_hostname=config["ENVOY_HOSTNAME"],
+                endpoint=endpoint,
+                token=credentials["token"],
+            )
+        )
+    )
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        sys.stderr.write(
+            f"""
+Usage:
+    {__file__} <endpoint>
+
+"""
+        )
+        sys.exit(1)
+
+    try:
+        main(endpoint=sys.argv[1])
+    except:
+        sys.stderr.write(traceback.format_exc())
+        sys.exit(1)

panel-json-to-csv.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+
+""" panel-json-to-csv.py
+
+Read a Envoy panel.json file and write a CSV version of it.
+
+This can be helpful when assigning row/col labels for your panels such as what
+is used by Envoy Logger.
+
+"""
+
+import csv
+import json
+from collections import OrderedDict
+
+with open("panel.json", "r") as f:
+    panel_json = json.load(f)
+
+config = {}
+
+for module in panel_json["arrays"][0]["modules"]:
+    config[module["inverter"]["serial_num"]] = {
+        "tags": {
+            "module_id": module["module_id"],
+            "x": module["x"],
+            "y": module["y"],
+            "inverter_id": module["inverter"]["inverter_id"],
+        },
+    }
+
+# Sort by x then y
+# This returns a list of tuples for some reason
+sorted_config = sorted(
+    config.items(), key=lambda i: (i[1]["tags"]["y"], i[1]["tags"]["x"])
+)
+
+config = OrderedDict()
+
+for c in sorted_config:
+    config[c[0]] = c[1]
+
+with open("panel.csv", "w") as f:
+    c = csv.writer(f)
+    c.writerow(["Serial", "X", "Y"])
+    for panel in config.items():
+        c.writerow([f"'{panel[0]}", panel[1]["tags"]["x"], panel[1]["tags"]["y"]])

panel-json-to-yaml.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+
+""" panel-json-to-yaml.py
+
+Read a Envoy panel.json file and write YAML version of it.
+
+This can be helpful when generating an Envoy Logger configuration file.
+
+"""
+
+import json
+from collections import OrderedDict
+
+import yaml
+
+with open("panel.json", "r") as f:
+    panel_json = json.load(f)
+
+config = {}
+
+for module in panel_json["arrays"][0]["modules"]:
+    config[module["inverter"]["serial_num"]] = {
+        "tags": {
+            "module_id": module["module_id"],
+            "x": module["x"],
+            "y": module["y"],
+            "inverter_id": module["inverter"]["inverter_id"],
+        },
+    }
+
+# Sort by x then y
+# This returns a list of tuples for some reason
+sorted_config = sorted(
+    config.items(), key=lambda i: (i[1]["tags"]["y"], i[1]["tags"]["x"])
+)
+
+config = OrderedDict()
+
+for c in sorted_config:
+    config[c[0]] = c[1]
+
+print(yaml.dump(config))

requirements.txt

@@ -0,0 +1,3 @@
+PyJWT
+python-dotenv
+requests

sample-data/README.md

@@ -0,0 +1,37 @@
+Sample data gathered From an Enphase Envoy/Gateway using the `envoy.py` script
+in the parent directory.  This data is from a site in the US with:
+
+- Single phase service (split phase / standard US 120v/240v)
+- Solar panels
+- Batteries
+- Generator
+- Central US timezone
+
+Timestamps appear to be in Unix format (seconds since midnight Jan 1, 1970).
+
+This data was dumped at night, so much of the production information is 0.
+
+Some data has been redacted as I'm unsure what is considered "private" in the
+Enphase world.  Where it has, you will find "REDACTED-\<description\>".  Some data
+is tracked through its redactions.  For example, a file with:
+
+    [
+        {
+            "some_secret_number": 12345,
+            "another_secret_number": 67890
+        }
+    ]
+
+Would be updated to:
+
+    [
+        {
+            "some_secret_number": REDACTED-5-DIGIT-INTEGER-A,
+            "another_secret_number": REDACTED-5-DIGIT-INTEGER-B
+        }
+    ]
+
+Note the lack of quotes (rendering the JSON invalid) because it was an integer
+and that the use of "A" and "B" in the name to indicate that they were
+different values.  Nested values may have "AA", "AB", "AC"... to indicate three
+unique values nested under previously redacted value "A".