Merge pull request #586 from smartenergycontrol-be/feature/influxdb-integration

Add InfluxDB integration for enhanced historical data retrieval

Author: davidusb-geek

docs/config.md

@@ -24,6 +24,14 @@ We will need to define these parameters to retrieve data from Home Assistant. Th
 	- 'sensor.power_load_no_var_loads'
 - `method_ts_round`: Set the method for timestamp rounding, options are: first, last and nearest.
 - `continual_publish`: set to True to save entities to .json after an optimization run. Then automatically republish the saved entities *(with updated current state value)* every `optimization_time_step` minutes. *entity data saved to data_path/entities.*
+- `use_influxdb`: Enable InfluxDB as a data source instead of the Home Assistant API. This allows for longer historical data retention and better performance for machine learning models.
+- `influxdb_host`: The IP address or hostname of your InfluxDB instance. Defaults to `localhost`.
+- `influxdb_port`: The port number for your InfluxDB instance. Defaults to 8086.
+- `influxdb_username`: Username for authenticating with InfluxDB. Leave empty if no authentication is required.
+- `influxdb_password`: Password for authenticating with InfluxDB. Leave empty if no authentication is required.
+- `influxdb_database`: The name of the InfluxDB database containing your Home Assistant data. Defaults to `homeassistant`.
+- `influxdb_measurement`: The measurement name where your sensor data is stored. Defaults to `W` for the Home Assistant integration.
+- `influxdb_retention_policy`: The retention policy to use for InfluxDB queries. Defaults to `autogen`.
 
 A second part of this section is given by some privacy-sensitive parameters that should be included in a `secrets_emhass.yaml` file alongside the `config_emhass.yaml` file.
 

docs/differences.md

@@ -24,6 +24,14 @@ See below for a list of associations between the parameters from `config_emhass.
 | retrieve_hass_conf |  set_zero_min |  set_zero_min | |
 | retrieve_hass_conf |  method_ts_round |  method_ts_round | |
 | retrieve_hass_conf |  continual_publish |  continual_publish | |
+| retrieve_hass_conf | | use_influxdb | |
+| retrieve_hass_conf | | influxdb_host | |
+| retrieve_hass_conf | | influxdb_port | |
+| retrieve_hass_conf | | influxdb_username | |
+| retrieve_hass_conf | | influxdb_password | |
+| retrieve_hass_conf | | influxdb_database | |
+| retrieve_hass_conf | | influxdb_measurement | |
+| retrieve_hass_conf | | influxdb_retention_policy | |
 | params_secrets |  solcast_api_key |  optional_solcast_api_key | |
 | params_secrets |  solcast_rooftop_id |  optional_solcast_rooftop_id | |
 | params_secrets |  solar_forecast_kwp |  optional_solar_forecast_kwp | |
@@ -79,6 +87,12 @@ Descriptions of each parameter can be found at:
 -  [`Configuration Documentation`](https://emhass.readthedocs.io/en/latest/config.html) 
 - Configuration page on EMHASS web server (E.g. http://localhost:5000/configuration)
 
+## InfluxDB as a data source
+A new feature allows using **InfluxDB** as an alternative data source to the Home Assistant recorder database. This is beneficial for users who want to treat longer data retention periods for training machine learning models or to reduce the query load on their main Home Assistant instance.
+
+When `use_influxdb: true` is set, EMHASS will fetch sensor data directly from your InfluxDB instance using the provided connection parameters. The `influxdb_username` and `influxdb_password` are treated as secrets.
+
+
 ## Passing in secret parameters
 Secret parameters are passed differently, depending on which method you choose. Alternative options are also present for passing secrets, if you are running EMHASS separately from Home Assistant. _(I.e. not via EMHASS-Add-on)_ 
 

pyproject.toml

@@ -38,6 +38,7 @@ dependencies = [
     "waitress>=3.0.2",
     "plotly>=6.0.0",
     "gunicorn>=23.0.0",
+    "influxdb>=5.3.1",
 ]
 [build-system]
 requires = ["hatchling"]

src/emhass/data/associations.csv

@@ -8,6 +8,14 @@ retrieve_hass_conf,load_negative,load_negative
 retrieve_hass_conf,set_zero_min,set_zero_min
 retrieve_hass_conf,var_replace_zero,sensor_replace_zero,list_sensor_replace_zero
 retrieve_hass_conf,var_interp,sensor_linear_interp,list_sensor_linear_interp
+retrieve_hass_conf,use_influxdb,use_influxdb
+retrieve_hass_conf,influxdb_host,influxdb_host
+retrieve_hass_conf,influxdb_port,influxdb_port
+retrieve_hass_conf,influxdb_username,influxdb_username
+retrieve_hass_conf,influxdb_password,influxdb_password
+retrieve_hass_conf,influxdb_database,influxdb_database
+retrieve_hass_conf,influxdb_measurement,influxdb_measurement
+retrieve_hass_conf,influxdb_retention_policy,influxdb_retention_policy
 retrieve_hass_conf,method_ts_round,method_ts_round
 retrieve_hass_conf,continual_publish,continual_publish
 params_secrets,time_zone,time_zone

src/emhass/data/config_defaults.json

@@ -29,6 +29,14 @@
     "sensor.power_photovoltaics",
     "sensor.power_load_no_var_loads"
   ],
+  "use_influxdb": false,
+  "influxdb_host": "localhost",
+  "influxdb_port": 8086,
+  "influxdb_username": "",
+  "influxdb_password": "",
+  "influxdb_database": "homeassistant",
+  "influxdb_measurement": "W",
+  "influxdb_retention_policy": "autogen",
   "load_negative": false,
   "set_zero_min": true,
   "number_of_deferrable_loads": 2,

src/emhass/retrieve_hass.py

@@ -78,6 +78,21 @@ def __init__(
         self.get_data_from_file = get_data_from_file
         self.var_list = []
 
+        # Initialize InfluxDB configuration
+        self.use_influxdb = self.params.get("retrieve_hass_conf", {}).get("use_influxdb", False)
+        if self.use_influxdb:
+            influx_conf = self.params.get("retrieve_hass_conf", {})
+            self.influxdb_host = influx_conf.get("influxdb_host", "localhost")
+            self.influxdb_port = influx_conf.get("influxdb_port", 8086)
+            self.influxdb_username = influx_conf.get("influxdb_username", "")
+            self.influxdb_password = influx_conf.get("influxdb_password", "")
+            self.influxdb_database = influx_conf.get("influxdb_database", "homeassistant")
+            self.influxdb_measurement = influx_conf.get("influxdb_measurement", "W")
+            self.influxdb_retention_policy = influx_conf.get("influxdb_retention_policy", "autogen")
+            self.logger.info(f"InfluxDB integration enabled: {self.influxdb_host}:{self.influxdb_port}/{self.influxdb_database}")
+        else:
+            self.logger.debug("InfluxDB integration disabled, using Home Assistant API")
+
     def get_ha_config(self):
         """
         Extract some configuration data from HA.
@@ -142,6 +157,10 @@ def get_data(
         .. warning:: The minimal_response and significant_changes_only options \
             are experimental
         """
+        # Use InfluxDB if configured, otherwise use Home Assistant API
+        if self.use_influxdb:
+            return self.get_data_influxdb(days_list, var_list)
+
         self.logger.info("Retrieve hass get data method initiated...")
         headers = {
             "Authorization": "Bearer " + self.long_lived_token,
@@ -301,6 +320,260 @@ def get_data(
         self.var_list = var_list
         return True
 
+    def get_data_influxdb(
+        self,
+        days_list: pd.date_range,
+        var_list: list,
+    ) -> bool:
+        """
+        Retrieve data from InfluxDB database.
+
+        This method provides an alternative data source to Home Assistant API,
+        enabling longer historical data retention for better machine learning model training.
+
+        :param days_list: A list of days to retrieve data for
+        :type days_list: pandas.date_range
+        :param var_list: List of sensor entity IDs to retrieve
+        :type var_list: list
+        :return: Success status of data retrieval
+        :rtype: bool
+        """
+        self.logger.info("Retrieve InfluxDB get data method initiated...")
+
+        # Check for empty inputs
+        if not days_list.size:
+            self.logger.error("Empty days_list provided")
+            return False
+
+        client = self._init_influx_client()
+        if not client:
+            return False
+
+        start_time = days_list[0]
+        end_time = days_list[-1] + pd.Timedelta(days=1)
+        total_days = (end_time - start_time).days
+
+        self.logger.info(f"Retrieving {len(var_list)} sensors over {total_days} days from InfluxDB")
+        self.logger.debug(f"Time range: {start_time} to {end_time}")
+
+        # Collect sensor dataframes
+        sensor_dfs = []
+        global_min_time = None
+        global_max_time = None
+
+        for sensor in filter(None, var_list):
+            df_sensor = self._fetch_sensor_data(client, sensor, start_time, end_time)
+            if df_sensor is not None:
+                sensor_dfs.append(df_sensor)
+                # Track global time range
+                sensor_min = df_sensor.index.min()
+                sensor_max = df_sensor.index.max()
+                global_min_time = min(global_min_time or sensor_min, sensor_min)
+                global_max_time = max(global_max_time or sensor_max, sensor_max)
+
+        client.close()
+
+        if not sensor_dfs:
+            self.logger.error("No data retrieved from InfluxDB")
+            return False
+
+        # Create complete time index covering all sensors
+        if global_min_time is not None and global_max_time is not None:
+            complete_index = pd.date_range(
+                start=global_min_time,
+                end=global_max_time,
+                freq=self.freq
+            )
+            self.df_final = pd.DataFrame(index=complete_index)
+
+            # Merge all sensor dataframes
+            for df_sensor in sensor_dfs:
+                self.df_final = pd.concat([self.df_final, df_sensor], axis=1)
+
+        # Set frequency and validate with error handling
+        try:
+            self.df_final = set_df_index_freq(self.df_final)
+        except Exception as e:
+            self.logger.error(f"Exception occurred while setting DataFrame index frequency: {e}")
+            return False
+
+        if self.df_final.index.freq != self.freq:
+            self.logger.warning(
+                f"InfluxDB data frequency ({self.df_final.index.freq}) differs from expected ({self.freq})"
+            )
+
+        self.var_list = var_list
+        self.logger.info(f"InfluxDB data retrieval completed: {self.df_final.shape}")
+        return True
+
+    def _init_influx_client(self):
+        """Initialize InfluxDB client connection."""
+        try:
+            from influxdb import InfluxDBClient
+        except ImportError:
+            self.logger.error("InfluxDB client not installed. Install with: pip install influxdb")
+            return None
+
+        try:
+            client = InfluxDBClient(
+                host=self.influxdb_host,
+                port=self.influxdb_port,
+                username=self.influxdb_username or None,
+                password=self.influxdb_password or None,
+                database=self.influxdb_database
+            )
+            # Test connection
+            client.ping()
+            self.logger.debug(f"Successfully connected to InfluxDB at {self.influxdb_host}:{self.influxdb_port}")
+
+            # Initialize measurement cache
+            if not hasattr(self, '_measurement_cache'):
+                self._measurement_cache = {}
+
+            return client
+        except Exception as e:
+            self.logger.error(f"Failed to connect to InfluxDB: {e}")
+            return None
+
+    def _discover_entity_measurement(self, client, entity_id: str) -> str:
+        """Auto-discover which measurement contains the given entity."""
+        # Check cache first
+        if entity_id in self._measurement_cache:
+            return self._measurement_cache[entity_id]
+
+        try:
+            # Get all available measurements
+            measurements_query = "SHOW MEASUREMENTS"
+            measurements_result = client.query(measurements_query)
+            measurements = [m['name'] for m in measurements_result.get_points()]
+
+            # Priority order: check common sensor types first
+            priority_measurements = ['EUR/kWh', '€/kWh', 'W', 'EUR', '€', '%', 'A', 'V']
+            all_measurements = priority_measurements + [m for m in measurements if m not in priority_measurements]
+
+            self.logger.debug(f"Searching for entity '{entity_id}' across {len(measurements)} measurements")
+
+            # Search for entity in each measurement
+            for measurement in all_measurements:
+                if measurement not in measurements:
+                    continue  # Skip if measurement doesn't exist
+
+                try:
+                    # Use SHOW TAG VALUES to get all entity_ids in this measurement
+                    tag_query = f'SHOW TAG VALUES FROM "{measurement}" WITH KEY = "entity_id"'
+                    self.logger.debug(f"Checking measurement '{measurement}' with tag query: {tag_query}")
+                    result = client.query(tag_query)
+                    points = list(result.get_points())
+
+                    # Check if our target entity_id is in the tag values
+                    for point in points:
+                        if point.get('value') == entity_id:
+                            self.logger.debug(f"Found entity '{entity_id}' in measurement '{measurement}'")
+                            # Cache the result
+                            self._measurement_cache[entity_id] = measurement
+                            return measurement
+
+                except Exception as query_error:
+                    self.logger.debug(f"Tag query failed for measurement '{measurement}': {query_error}")
+                    continue
+
+        except Exception as e:
+            self.logger.error(f"Error discovering measurement for entity {entity_id}: {e}")
+
+        # Fallback to default measurement if not found
+        self.logger.warning(f"Entity '{entity_id}' not found in any measurement, using default: {self.influxdb_measurement}")
+        return self.influxdb_measurement
+
+    def _build_influx_query_for_measurement(self, entity_id: str, measurement: str, start_time, end_time) -> str:
+        """Build InfluxQL query for specific measurement and entity."""
+        # Convert frequency to InfluxDB interval
+        freq_minutes = int(self.freq.total_seconds() / 60)
+        interval = f"{freq_minutes}m"
+
+        # Format times properly for InfluxDB
+        start_time_str = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
+        end_time_str = end_time.strftime('%Y-%m-%dT%H:%M:%SZ')
+
+        # Use FILL(previous) instead of FILL(linear) for compatibility with open-source InfluxDB
+        query = f'''
+        SELECT mean("value") AS "mean_value"
+        FROM "{self.influxdb_database}"."{self.influxdb_retention_policy}"."{measurement}"
+        WHERE time >= '{start_time_str}'
+        AND time < '{end_time_str}'
+        AND "entity_id"='{entity_id}'
+        GROUP BY time({interval}) FILL(previous)
+        '''
+        return query
+
+    def _build_influx_query(self, sensor: str, start_time, end_time) -> str:
+        """Build InfluxQL query for sensor data retrieval (legacy method)."""
+        # Convert sensor name: sensor.sec_pac_solar -> sec_pac_solar
+        entity_id = sensor.replace('sensor.', '') if sensor.startswith('sensor.') else sensor
+
+        # Use default measurement (for backward compatibility)
+        return self._build_influx_query_for_measurement(entity_id, self.influxdb_measurement, start_time, end_time)
+
+    def _fetch_sensor_data(self, client, sensor: str, start_time, end_time):
+        """Fetch and process data for a single sensor with auto-discovery."""
+        self.logger.debug(f"Retrieving sensor: {sensor}")
+
+        # Clean sensor name (remove sensor. prefix if present)
+        entity_id = sensor.replace('sensor.', '') if sensor.startswith('sensor.') else sensor
+
+        # Auto-discover which measurement contains this entity
+        measurement = self._discover_entity_measurement(client, entity_id)
+        if not measurement:
+            self.logger.warning(f"Entity '{entity_id}' not found in any InfluxDB measurement")
+            return None
+
+        try:
+            query = self._build_influx_query_for_measurement(entity_id, measurement, start_time, end_time)
+            self.logger.debug(f"InfluxDB query: {query}")
+
+            # Execute query
+            result = client.query(query)
+            points = list(result.get_points())
+
+            if not points:
+                self.logger.warning(f"No data found for entity: {entity_id} in measurement: {measurement}")
+                return None
+
+            self.logger.info(f"Retrieved {len(points)} data points for {sensor}")
+
+            # Create DataFrame from points
+            df_sensor = pd.DataFrame(points)
+
+            # Convert time column and set as index with timezone awareness
+            df_sensor['time'] = pd.to_datetime(df_sensor['time'], utc=True)
+            df_sensor.set_index('time', inplace=True)
+
+            # Rename value column to original sensor name
+            if 'mean_value' in df_sensor.columns:
+                df_sensor = df_sensor[['mean_value']].rename(columns={'mean_value': sensor})
+            else:
+                self.logger.error(f"Expected 'mean_value' column not found for {sensor} in measurement {measurement}")
+                return None
+
+            # Handle non-numeric data with NaN ratio warning
+            df_sensor[sensor] = pd.to_numeric(df_sensor[sensor], errors='coerce')
+
+            # Check proportion of NaNs and log warning if high
+            nan_count = df_sensor[sensor].isna().sum()
+            total_count = len(df_sensor[sensor])
+            if total_count > 0:
+                nan_ratio = nan_count / total_count
+                if nan_ratio > 0.2:
+                    self.logger.warning(
+                        f"Entity '{entity_id}' has {nan_count}/{total_count} ({nan_ratio:.1%}) non-numeric values coerced to NaN."
+                    )
+
+            self.logger.debug(f"Successfully retrieved {len(df_sensor)} data points for '{entity_id}' from measurement '{measurement}'")
+            return df_sensor
+
+        except Exception as e:
+            self.logger.error(f"Failed to query entity {entity_id} from measurement {measurement}: {e}")
+            return None
+
     def prepare_data(
         self,
         var_load: str,