Merge pull request #117 from aryanbhosale/main

auth grant flow

.gitignore

@@ -2,5 +2,6 @@ scripts/datapipes/test.nc
 scripts/datapipes/UK_PV_metadata.csv
 scripts/datapipes/nwp_data
 quartz_solar_forecast/data
+quartz_solar_forecast.egg-info
 .env
 venv

examples/inverter_example.py

@@ -0,0 +1,18 @@
+""" Example code to run the forecast"""
+from quartz_solar_forecast.forecast import run_forecast
+from quartz_solar_forecast.pydantic_models import PVSite
+from datetime import datetime, timedelta
+
+def main():
+    # make input data
+    site = PVSite(latitude=51.75, longitude=-1.25, capacity_kwp=1.25, inverter_type="enphase")
+
+    ts = datetime.today() - timedelta(weeks=1)
+    predictions_df = run_forecast(site=site, ts=ts, nwp_source="icon")
+
+    print(predictions_df)
+    print(f"Max: {predictions_df['power_wh'].max()}")
+
+
+if __name__ == "__main__":
+    main()

quartz_solar_forecast/data.py

@@ -21,7 +21,7 @@
 
 from dotenv import load_dotenv
 
-ENPHASE_SYSTEM_ID = os.getenv('ENPHASE_SYSTEM_ID')
+system_id = os.getenv('ENPHASE_SYSTEM_ID')
 
 def get_nwp(site: PVSite, ts: datetime, nwp_source: str = "icon") -> xr.Dataset:
     """
@@ -165,7 +165,7 @@ def make_pv_data(site: PVSite, ts: pd.Timestamp) -> xr.Dataset:
             site_id = matching_site_ids[0]
             live_generation_wh = get_solaredge_data(site_id)
     elif site.inverter_type == 'enphase':
-        live_generation_wh = get_enphase_data(ENPHASE_SYSTEM_ID)
+        live_generation_wh = get_enphase_data(system_id)
     else:
         # If no inverter type is specified, use the default value
         live_generation_wh = np.nan

quartz_solar_forecast/inverters/enphase.py

@@ -1,47 +1,139 @@
 import requests
+import http.client
 import os
+import json
+import base64
+from datetime import datetime, timedelta
 
 from dotenv import load_dotenv
 
-ENPHASE_CLIENT_ID = os.getenv('ENPHASE_CLIENT_ID')
-ENPHASE_CLIENT_SECRET = os.getenv('ENPHASE_CLIENT_SECRET')
-ENPHASE_API_KEY = os.getenv('ENPHASE_API_KEY')
+import os
+from urllib.parse import urlencode
+
+def get_enphase_auth_url():
+    """
+    Generate the authorization URL for the Enphase API.
+
+    :param None
+    :return: Authentication URL
+    """
+    client_id = os.getenv('ENPHASE_CLIENT_ID')
+
+    redirect_uri = (
+        "https://api.enphaseenergy.com/oauth/redirect_uri"  # Or your own redirect URI
+    )
+    params = {
+        "response_type": "code",
+        "client_id": client_id,
+        "redirect_uri": redirect_uri,
+    }
+    auth_url = f"https://api.enphaseenergy.com/oauth/authorize?{urlencode(params)}"
+    return auth_url
+
+
+def get_enphase_authorization_code(auth_url):
+    """
+    Open the authorization URL in a browser and retrieve the authorization code from the redirect URI.
+
+    :param auth_url: Authentication URL to get the code
+    :return: The one time code for access to a system
+    """
+    # Open the authorization URL in a browser
+    print(f"Please visit the following URL and authorize the application: {auth_url}")
+    print(
+        "After authorization, you will be redirected to a URL with the authorization code."
+    )
+    print("Please copy and paste the full redirect URL here:")
+    redirect_url = input()
+    # Extract the authorization code from the redirect URL
+    code = redirect_url.split("?code=")[1]
+    return code
+
 
 def get_enphase_access_token():
     """
-    Obtain an access token for the Enphase API using the Client Credentials Grant flow.
+    Obtain an access token for the Enphase API using the Authorization Code Grant flow.
+    :param None
+    :return: Access Token
     """
-    url = "https://api.enphaseenergy.com/oauth/token"
+
+    client_id = os.getenv('ENPHASE_CLIENT_ID')
+    client_secret = os.getenv('ENPHASE_CLIENT_SECRET')
+
+    auth_url = get_enphase_auth_url()
+    auth_code = get_enphase_authorization_code(auth_url)
+
+    # Combine the client ID and secret with a colon separator
+    credentials = f"{client_id}:{client_secret}"
+
+    # Encode the credentials as bytes
+    credentials_bytes = credentials.encode("utf-8")
+
+    # Base64 encode the bytes
+    encoded_credentials = base64.b64encode(credentials_bytes)
+
+    # Convert the encoded bytes to a string
+    encoded_credentials_str = encoded_credentials.decode("utf-8")
+
+    conn = http.client.HTTPSConnection("api.enphaseenergy.com")
+    payload = ""
     headers = {
-        "Content-Type": "application/x-www-form-urlencoded",
-        "Authorization": f"Basic {(ENPHASE_CLIENT_ID + ':' + ENPHASE_CLIENT_SECRET).encode().decode('utf-8')}",
-    }
-    data = {
-        "grant_type": "client_credentials",
-        "scope": "read"
+        "Authorization": f"Basic {encoded_credentials_str}"
     }
+    conn.request(
+        "POST",
+        f"/oauth/token?grant_type=authorization_code&redirect_uri=https://api.enphaseenergy.com/oauth/redirect_uri&code={auth_code}",
+        payload,
+        headers,
+    )
+    res = conn.getresponse()
+    data = res.read()
+
+    # Decode the data read from the response
+    decoded_data = data.decode("utf-8")
+
+    # Convert the decoded data into JSON format
+    data_json = json.loads(decoded_data)
+    access_token = data_json["access_token"]
+
+    return access_token
 
-    response = requests.post(url, headers=headers, data=data)
-    response.raise_for_status()
-    return response.json()["access_token"]
 
 def get_enphase_data(enphase_system_id: str) -> float:
-    """
+    """ 
     Get live PV generation data from Enphase API v4
-
     :param enphase_system_id: System ID for Enphase API
     :return: Live PV generation in Watt-hours, assumes to be a floating-point number
     """
-    url = f'https://api.enphaseenergy.com/api/v4/{enphase_system_id}/summary'
+    api_key = os.getenv('ENPHASE_API_KEY')
+    access_token = get_enphase_access_token()
+
+    # Set the start time to 30mins from now
+    start_at = int((datetime.now() - timedelta(minutes=30)).timestamp())
+
+    # Set the granularity to day
+    granularity = "day"
+
+    conn = http.client.HTTPSConnection("api.enphaseenergy.com")
     headers = {
-        'Authorization': f'Bearer {get_enphase_access_token()}',
-        'key': ENPHASE_API_KEY
+        "Authorization": f"Bearer {str(access_token)}",
+        "key": str(api_key)
     }
 
-    response = requests.get(url, headers=headers)
-    data = response.json()
+    # Add the system_id and duration parameters to the URL
+    url = f"/api/v4/systems/{enphase_system_id}/telemetry/production_micro?start_at={start_at}&granularity={granularity}"
+    conn.request("GET", url, headers=headers)
+
+    res = conn.getresponse()
+    data = res.read()
+
+    # Decode the data read from the response
+    decoded_data = data.decode("utf-8")
+
+    # Convert the decoded data into JSON format
+    data_json = json.loads(decoded_data)
 
+    ### TO-DO
     # Extracting live generation data assuming it's in Watt-hours
-    live_generation_wh = data['current_power']['power']
-
+    live_generation_wh = data_json['current_power']['power']
     return live_generation_wh