Test Python

voyanti_ems/config.yaml

@@ -16,11 +16,10 @@ map:
 options:
   message: "Hello world.. VY!"
   VOYANTI_SITE_ID: "xxxxxxxxxxxxx"
-  mqtt:
-    - MQTT_HOST: str
-      MQTT_PORT: port
-      MQTT_USERNAME: str
-      MQTT_PASSWORD: password
+  MQTT_HOST: str
+  MQTT_PORT: port
+  MQTT_USERNAME: str
+  MQTT_PASSWORD: password
 schema:
   message: "str?"
 image: "ghcr.io/heinrich321/{arch}-voyanti-ems"

voyanti_ems/rootfs/usr/src/app.py

@@ -1,6 +1,7 @@
 import time
 import logging
 import requests
+from solar import fetch_solar_forecast, publish_solar_forecast
 
 # Setup basic logging
 logging.basicConfig(level=logging.INFO)
@@ -28,6 +29,18 @@ def fetch_data():
 if __name__ == "__main__":
     logger.info("Starting the Python data-fetching script...")
 
+    forecast_data = fetch_solar_forecast()
+
+    logger.info(f"Solar Data: {forecast_data}")
+
+    if forecast_data:
+        # Extract relevant data
+        total_today_kwh = forecast_data.get("total_today_kwh", 0)
+        day_profile = forecast_data.get("day_profile", [])
+
+        # Publish the forecast data to Home Assistant, including time series data
+        publish_solar_forecast(total_today_kwh, day_profile)
+
     # Infinite loop to fetch data every 60 seconds
     while True:
         fetch_data()

voyanti_ems/rootfs/usr/src/solar.py

@@ -0,0 +1,62 @@
+import os
+import requests
+import json
+
+# Home Assistant Supervisor API Configuration
+HOME_ASSISTANT_URL = "http://supervisor/core/api"
+SUPERVISOR_TOKEN = os.getenv("SUPERVISOR_TOKEN")  # Access the token from the environment variable
+
+# VOYANTI Site ID (passed as add-on option)
+VOYANTI_SITE_ID = os.getenv("VOYANTI_SITE_ID")  # Replace with default or add-on configuration
+API_URL = f"https://xi1qbxerg8.execute-api.eu-north-1.amazonaws.com/prod/site-forecast?siteId={VOYANTI_SITE_ID}"
+
+# Headers for Home Assistant API requests
+HEADERS = {
+    "Authorization": f"Bearer {SUPERVISOR_TOKEN}",
+    "Content-Type": "application/json"
+}
+
+# Home Assistant sensor entity to update
+UPDATE_SENSOR_URL = f"{HOME_ASSISTANT_URL}/states/sensor.solar_forecast"
+
+# Function to fetch solar forecast from the API
+def fetch_solar_forecast():
+    try:
+        response = requests.get(API_URL)
+        if response.status_code == 200:
+            return response.json()
+        else:
+            print(f"Failed to fetch data from API: {response.status_code}")
+            return None
+    except Exception as e:
+        print(f"Error fetching data: {e}")
+        return None
+
+# Function to publish solar forecast to Home Assistant, including time series data
+def publish_solar_forecast(total_today_kwh, day_profile):
+    # Create a dictionary to hold the time series forecast
+    time_series_forecast = {entry["period_end"]: entry["pv_energy_rooftop"] for entry in day_profile}
+
+    # Create the payload with time series data as attributes
+    payload = {
+        "state": total_today_kwh,  # Total forecasted kWh for today
+        "attributes": {
+            "unit_of_measurement": "kWh",
+            "friendly_name": "Voyanti Solar Energy Forecast",
+            "device_class": "energy",
+            "state_class": "measurement",
+            "forecast": True,
+            "time_series_forecast": time_series_forecast  # Add time series forecast as an attribute
+        }
+    }
+
+    try:
+        # Publish the forecast data to Home Assistant
+        response = requests.post(UPDATE_SENSOR_URL, headers=HEADERS, data=json.dumps(payload))
+        if response.status_code == 200:
+            print("Solar forecast published successfully.")
+        else:
+            print(f"Failed to publish forecast: {response.status_code}")
+            print(response.text)
+    except Exception as e:
+        print(f"Error occurred while publishing forecast: {e}")