Add power data retrieval and database storage functionality

Implemented methods to fetch power data from the SEMS Portal API and write energy-related metrics to the database. Updated the inverter charge controller to call this functionality. Improved InfluxDB usage instructions in the README for better clarity and usage.

Author: felixschndr

README.md

@@ -213,14 +213,26 @@ It saves the following data:
 
 ## InfluxDB commands
 
-- Create bucket: `influx bucket create -org default -token <token> --name default`
-- Delete bucket: `influx bucket delete -org default -token <token> --name default`
-- Retrieve all solar forecast values: `influx query -org default -token <token> 'import "experimental"
+- Create bucket: `influx bucket create -org default -token ${TOKEN} --name default`
+- Delete bucket: `influx bucket delete -org default -token ${TOKEN} --name default`
+- Retrieve all solar forecast values: `influx query -org default -token ${TOKEN} 'import "experimental"
 from(bucket: "default")
 |> range(start: 0, stop: experimental.addDuration(d: 2d, to: now()))
 |> filter(fn: (r) => r._measurement == "sun_forecast")
 |> pivot(rowKey:["_time"], columnKey:["_field"], valueColumn:"_value")'`
-- Retrieve all energy prices: `influx query -org default -token <token> 'import "experimental"
+- Retrieve all energy prices: `influx query -org default -token ${TOKEN} 'import "experimental"
 from(bucket: "default")
 |> range(start: 0, stop: experimental.addDuration(d: 2d, to: now()))
 |> filter(fn: (r) => r._measurement == "energy_prices")'`
+- Retrieve all power data (semsportal): `influx query -org default -token ${TOKEN} 'import "experimental"
+from(bucket: "default")
+|> range(start: 0, stop: experimental.addDuration(d: 2d, to: now()))
+|> filter(fn: (r) => r._measurement == "power")
+|> pivot(rowKey:["_time"], columnKey:["_field"], valueColumn:"_value")'`
+- Copy data from one measurement to another: `influx query -org default -token ${TOKEN} 'import "experimental"
+from(bucket: "default")
+  |> range(start: 0, stop: experimental.addDuration(d: 2d, to: now()))
+  |> filter(fn: (r) => r._measurement == "<old_measurement>")
+  |> set(key: "_measurement", value: "<new_measurement>")
+  |> to(bucket: "default")'`
+- Delete data from one measurement: `influx delete --bucket default -org default -token ${TOKEN} --start='1970-01-01T00:00:00Z' --stop=$(date +"%Y-%m-%dT%H:%M:%SZ" -d "+2 days") --predicate '_measurement=<old_measurement>'`

source/inverter_charge_controller.py

@@ -75,6 +75,8 @@ def _start(self) -> None:
         duration_to_wait_in_cause_of_error = timedelta(minutes=10)
         while True:
             try:
+                self.sems_portal_api_handler.write_values_to_database()
+
                 if first_iteration:
                     next_price_minimum = self.tibber_api_handler.get_next_price_minimum(first_iteration)
                     first_iteration = False

source/sems_portal_api_handler.py

@@ -1,6 +1,7 @@
-from datetime import datetime, time
+from datetime import date, datetime, time, timedelta
 
 import requests
+from database_handler import DatabaseHandler, InfluxDBField
 from energy_amount import EnergyAmount, Power
 from environment_variable_getter import EnvironmentVariableGetter
 from logger import LoggerMixin
@@ -16,6 +17,8 @@ def __init__(self):
         self.timestamp = None
         self.user_id = None
 
+        self.database_handler = DatabaseHandler("power")
+
     def login(self) -> None:
         """
         Authenticates a user by sending a POST request to the SEMS Portal API and retrieves
@@ -63,15 +66,51 @@ def get_average_energy_consumption_per_day(self) -> EnergyAmount:
         """
         self.log.debug("Determining average energy consumption per day")
 
-        self.login()
-
         api_response = self._retrieve_energy_consumption_data()
         consumption_data = self._extract_energy_usage_data_of_response(api_response)
         average_consumption_per_day = sum([consumption.watt_hours for consumption in consumption_data]) / len(
             consumption_data
         )
         return EnergyAmount(watt_hours=average_consumption_per_day)
 
+    def _retrieve_power_data(self, date_to_crawl: date) -> dict:
+        """
+        Retrieves the power data from the SEMSPORTAL API. This includes:
+         - solar generation
+         - battery charge/discharge
+         - grid consumption/feed
+         - power usage
+         - state of charge
+
+        This method sends a POST request to the SEMSPORTAL API to fetch the energy consumption data of a specified plant station.
+        It constructs the necessary headers and payload required by the API and handles the response appropriately.
+
+        Returns:
+            dict: A dictionary containing the power data retrieved from the SEMSPORTAL API.
+        """
+        self.login()
+
+        self.log.debug("Crawling the SEMSPORTAL API for power data...")
+
+        url = "https://eu.semsportal.com/api/v2/Charts/GetPlantPowerChart"
+        headers = {
+            "Content-Type": "application/json",
+            "Token": f'{{"version":"v2.1.0","client":"ios","language":"en", "timestamp": "{self.timestamp}", "uid": "{self.user_id}", "token": "{self.token}"}}',
+        }
+        payload = {
+            "id": EnvironmentVariableGetter.get("SEMSPORTAL_POWERSTATION_ID"),
+            "date": date_to_crawl.strftime("%Y-%m-%d"),
+            "full_script": False,
+        }
+
+        response = requests.post(url, headers=headers, json=payload, timeout=20)
+        response.raise_for_status()
+        response = response.json()
+
+        self.log.trace(f"Retrieved data: {response}")
+
+        return response
+
     def _retrieve_energy_consumption_data(self) -> dict:
         """
         Retrieves energy consumption data from the SEMSPORTAL API.
@@ -82,6 +121,8 @@ def _retrieve_energy_consumption_data(self) -> dict:
         Returns:
             dict: A dictionary containing the energy consumption data retrieved from the SEMSPORTAL API.
         """
+        self.login()
+
         self.log.debug("Crawling the SEMSPORTAL API for energy consumption data...")
 
         url = "https://eu.semsportal.com/api/v2/Charts/GetChartByPlant"
@@ -218,7 +259,64 @@ def estimate_energy_usage_in_timeframe(self, timestamp_start: datetime, timestam
 
         return energy_usage_during_the_day + energy_usage_during_the_night
 
+    def write_values_to_database(self) -> None:
+        """
+        Writes energy-related metrics to the database for the last three days.
+
+        This method retrieves and processes energy data for the past three days, including solar generation, battery
+        discharge, grid feed, power usage and state of charge. It writes the resulting records to the database.
+        """
+        self.log.debug("Writing values to database...")
+        today = date.today()
+        for days_in_past in range(3):
+            date_to_crawl = today - timedelta(days=days_in_past)
+            data = self._retrieve_power_data(date_to_crawl)
+            self.log.debug(f"Retrieved power data: {data}")
+            lines = data["data"]["lines"]
+
+            time_keys = [line["x"] for line in lines[0]["xy"]]
+            for time_key in time_keys:
+                timestamp = datetime.combine(date_to_crawl, datetime.strptime(time_key, "%H:%M").time())
+                timestamp = timestamp.replace(tzinfo=TimeHandler.get_timezone())
+                self.database_handler.write_to_database(
+                    [
+                        InfluxDBField(
+                            "solar_generation", self._get_value_of_line_by_line_index_and_time_key(lines, 0, time_key)
+                        ),
+                        InfluxDBField(
+                            "battery_discharge", self._get_value_of_line_by_line_index_and_time_key(lines, 1, time_key)
+                        ),
+                        InfluxDBField(
+                            "grid_feed", self._get_value_of_line_by_line_index_and_time_key(lines, 2, time_key)
+                        ),
+                        InfluxDBField(
+                            "power_usage", self._get_value_of_line_by_line_index_and_time_key(lines, 3, time_key)
+                        ),
+                        InfluxDBField(
+                            "state_of_charge", self._get_value_of_line_by_line_index_and_time_key(lines, 4, time_key)
+                        ),
+                    ],
+                    timestamp,
+                )
+
+    @staticmethod
+    def _get_value_of_line_by_line_index_and_time_key(lines: dict, line_index: int, time_key: str) -> int:
+        """
+        Retrieves the value associated with a specific time key from a line in a nested dictionary structure.
+
+        This method searches for a specific time key ('x') within the 'xy' list of dictionaries
+        contained in a given line at a specified index. It extracts the associated value ('y')
+        for the matching time key and converts it to an integer.
+
+        Args:
+            lines (dict): A dictionary where each key corresponds to a line index. Each line index
+                maps to a dictionary containing a key 'xy', which is a list of dictionaries.
+                Each dictionary within 'xy' contains keys 'x' and 'y'.
+            line_index (int): The index of the line to search within the 'lines' dictionary.
+            time_key (str): The time key to search for, used to identify the specific dictionary in
+                the 'xy' list where the 'x' value matches.
 
-if __name__ == "__main__":
-    s = SemsPortalApiHandler()
-    print(s.get_energy_buy(1))
+        Returns:
+            int: The integer representation of the value ('y') corresponding to the provided time key.
+        """
+        return int([line for line in lines[line_index]["xy"] if line["x"] == time_key][0]["y"])