API Reference

emhass.command_line module

class emhass.command_line.OptimizationCache

Bases: object

In-memory cache for Optimization objects to enable warm-starting.

Warm-starting reuses the previous solution as a starting point for the solver, which can significantly speed up repeated MPC optimizations where consecutive problems are similar.

The cache is invalidated when configuration changes that affect the optimization structure (number of variables, constraints, etc.).

Thread-safe: Uses a lock to prevent race conditions when multiple optimizations run concurrently in async code.

classmethod clear(logger: Logger | None = None) → None

Clear the cache (e.g., for testing or explicit invalidation). Thread-safe.

classmethod get(optim_conf: dict, plant_conf: dict, costfun: str, retrieve_hass_conf: dict, logger: Logger, num_timesteps: int | None = None) → Optimization | None

Get cached Optimization object if configuration matches.

Returns None if cache is empty or configuration has changed. Thread-safe via internal locking.

classmethod get_stats() → dict

Get cache statistics for debugging. Thread-safe.

classmethod put(opt: Optimization, optim_conf: dict, plant_conf: dict, costfun: str, retrieve_hass_conf: dict, logger: Logger, num_timesteps: int | None = None) → None

Store Optimization object in cache. Thread-safe via internal locking.

class emhass.command_line.OptimizationCacheKey(number_of_deferrable_loads: int, set_use_battery: bool, set_use_pv: bool, treat_deferrable_load_as_semi_cont: tuple, set_deferrable_load_single_constant: tuple, set_deferrable_startup_penalty: tuple, deferrable_load_max_cost: tuple, set_deferrable_max_startups: tuple, set_deferrable_load_as_timeseries: tuple, nominal_power_of_deferrable_loads: tuple, def_load_config_structure: tuple, deferrable_load_groups: tuple, shared_thermal_tanks: tuple, is_electric_load: tuple, inverter_is_hybrid: bool, compute_curtailment: bool, optimization_time_step_s: float | None, delta_forecast_daily_s: float | None, num_timesteps: int | None, costfun: str, plant_conf_hash: str, optim_conf_structural_hash: str)

Bases: object

Frozen dataclass representing configuration fields that affect optimization structure.

Changes to any of these fields require rebuilding the optimization problem. Using a frozen dataclass makes the cache key explicit, hashable, and easy to extend.

compute_curtailment: bool

costfun: str

def_load_config_structure: tuple

deferrable_load_groups: tuple

deferrable_load_max_cost: tuple

delta_forecast_daily_s: float | None

inverter_is_hybrid: bool

is_electric_load: tuple

nominal_power_of_deferrable_loads: tuple

num_timesteps: int | None

number_of_deferrable_loads: int

optim_conf_structural_hash: str

optimization_time_step_s: float | None

plant_conf_hash: str

set_deferrable_load_as_timeseries: tuple

set_deferrable_load_single_constant: tuple

set_deferrable_max_startups: tuple

set_deferrable_startup_penalty: tuple

set_use_battery: bool

set_use_pv: bool

shared_thermal_tanks: tuple

treat_deferrable_load_as_semi_cont: tuple

class emhass.command_line.PublishContext(input_data_dict: dict, params: dict, idx: int, common_kwargs: dict, logger: Logger)

Bases: object

Context object for data publishing helpers.

Attributes:

input_data_dict (dict): Dictionary containing input data with keys ‘rh’ (RetrieveHass),

‘opt’ (Optimization - may be None for publish-data), and ‘fcst’ (Forecast) objects.

params (dict): Parameters dictionary for publishing configuration. idx (int): Index identifier for the current publishing operation. common_kwargs (dict): Common keyword arguments shared across publishing helpers. logger (logging.Logger): Logger instance for recording publishing operations.

common_kwargs: dict

property fcst: Forecast

idx: int

input_data_dict: dict

logger: Logger

property opt: Optimization

property optim_conf: dict

Access optim_conf directly from input_data_dict (works even when opt is None).

params: dict

property plant_conf: dict

Access plant_conf directly from input_data_dict (works even when opt is None).

property rh: RetrieveHass

class emhass.command_line.SetupContext(retrieve_hass_conf: dict, optim_conf: dict, plant_conf: dict, emhass_conf: dict, params: dict, logger: Logger, get_data_from_file: bool, rh: RetrieveHass, fcst: Forecast | None = None)

Bases: object

A dataclass that serves as a context container for optimization preparation helpers. This context object encapsulates all necessary configuration and utility objects required for setting up and preparing optimization tasks.

Attributes:

retrieve_hass_conf (dict): Configuration dictionary for Home Assistant data retrieval. optim_conf (dict): Configuration dictionary for optimization parameters. plant_conf (dict): Configuration dictionary for plant/system parameters. emhass_conf (dict): Configuration dictionary for EMHASS settings. params (dict): Additional parameters dictionary. logger (logging.Logger): Logger instance for logging messages. get_data_from_file (bool): Flag indicating whether to retrieve data from file instead of live source. rh (RetrieveHass): RetrieveHass instance for retrieving Home Assistant data. fcst (Forecast | None): Optional Forecast object for weather or energy forecasting. Defaults to None.

emhass_conf: dict

fcst: Forecast | None = None

get_data_from_file: bool

logger: Logger

optim_conf: dict

params: dict

plant_conf: dict

retrieve_hass_conf: dict

rh: RetrieveHass

async emhass.command_line.adjust_pv_forecast(logger: Logger, fcst: Forecast, p_pv_forecast: Series, get_data_from_file: bool, retrieve_hass_conf: dict, optim_conf: dict, rh: RetrieveHass, emhass_conf: dict, test_df_literal: DataFrame) → Series

Adjust the photovoltaic (PV) forecast using historical data and a regression model.

This method retrieves historical data, prepares it for model fitting, trains a regression model, and adjusts the provided PV forecast based on the trained model.

Parameters:

• logger (logging.Logger) – Logger object for logging information and errors.

• fcst (Forecast) – Forecast object used for PV forecast adjustment.

• p_pv_forecast (pd.Series) – The initial PV forecast to be adjusted.

• get_data_from_file (bool) – Whether to retrieve data from a file instead of Home Assistant.

• retrieve_hass_conf (dict) – Configuration dictionary for retrieving data from Home Assistant.

• optim_conf (dict) – Configuration dictionary for optimization settings.

• rh (RetrieveHass) – RetrieveHass object for interacting with Home Assistant.

• emhass_conf (dict) – Configuration dictionary for emhass paths and settings.

• test_df_literal (pd.DataFrame) – DataFrame containing test data for debugging purposes.

Returns:

The adjusted PV forecast as a pandas Series.

Return type:

pd.Series

async emhass.command_line.continual_publish(input_data_dict: dict, entity_path: Path, logger: Logger)

If continual_publish is true and a entity file saved in /data_path/entities, continually publish sensor on freq rate, updating entity current state value based on timestamp

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• entity_path (Path) – Path for entities folder in data_path

• logger (logging.Logger) – The passed logger object

async emhass.command_line.dayahead_forecast_optim(input_data_dict: dict, logger: Logger, save_data_to_file: bool | None = False, debug: bool | None = False) → DataFrame

Perform a call to the day-ahead optimization routine.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging object) – The passed logger object

• save_data_to_file (bool, optional) – Save optimization results to CSV file

• debug (bool, optional) – A debug option useful for unittests

Returns:

The output data of the optimization

Return type:

pd.DataFrame

async emhass.command_line.export_influxdb_to_csv(input_data_dict: dict | None, logger: Logger, emhass_conf: dict | None = None, params: str | None = None, runtimeparams: str | None = None) → bool

Export data from InfluxDB to CSV file.

This function can be called in two ways: 1. With input_data_dict (from web_server via set_input_data_dict) 2. Without input_data_dict (direct call from command line or web_server before set_input_data_dict)

Parameters:

• input_data_dict (dict | None) – Dictionary containing configuration and parameters (optional)

• logger (logging.Logger) – Logger object

• emhass_conf (dict | None) – Dictionary containing EMHASS configuration paths (used when input_data_dict is None)

• params (str | None) – JSON string of params (used when input_data_dict is None)

• runtimeparams (str | None) – JSON string of runtime parameters (used when input_data_dict is None)

Returns:

Success status

Return type:

bool

async emhass.command_line.forecast_model_fit(input_data_dict: dict, logger: Logger, debug: bool | None = False) → tuple[DataFrame, DataFrame, MLForecaster]

Perform a forecast model fit from training data retrieved from Home Assistant.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging.Logger) – The passed logger object

• debug (Optional[bool], optional) – True to debug, useful for unit testing, defaults to False

Returns:

The DataFrame containing the forecast data results without and with backtest and the mlforecaster object

Return type:

Tuple[pd.DataFrame, pd.DataFrame, mlforecaster]

async emhass.command_line.forecast_model_predict(input_data_dict: dict, logger: Logger, use_last_window: bool | None = True, debug: bool | None = False, mlf: MLForecaster | None = None) → DataFrame

Perform a forecast model predict using a previously trained skforecast model.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging.Logger) – The passed logger object

• use_last_window (Optional[bool], optional) – True if the ‘last_window’ option should be used for the custom machine learning forecast model. The ‘last_window=True’ means that the data that will be used to generate the new forecast will be freshly retrieved from Home Assistant. This data is needed because the forecast model is an auto-regressive model with lags. If ‘False’ then the data using during the model train is used. Defaults to True

• debug (Optional[bool], optional) – True to debug, useful for unit testing, defaults to False

• mlf (Optional[mlforecaster], optional) – The ‘mlforecaster’ object previously trained. This is mainly used for debug and unit testing. In production the actual model will be read from a saved pickle file. Defaults to None

Returns:

The DataFrame containing the forecast prediction data

Return type:

pd.DataFrame

async emhass.command_line.forecast_model_tune(input_data_dict: dict, logger: Logger, debug: bool | None = False, mlf: MLForecaster | None = None) → tuple[DataFrame, MLForecaster]

Tune a forecast model hyperparameters using bayesian optimization.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging.Logger) – The passed logger object

• debug (Optional[bool], optional) – True to debug, useful for unit testing, defaults to False

• mlf (Optional[mlforecaster], optional) – The ‘mlforecaster’ object previously trained. This is mainly used for debug and unit testing. In production the actual model will be read from a saved pickle file. Defaults to None

Returns:

The DataFrame containing the forecast data results using the optimized model

Return type:

pd.DataFrame

emhass.command_line.is_model_outdated(model_path: Path, max_age_hours: int, logger: Logger) → bool

Check if the saved model file is outdated based on its modification time.

Parameters:

• model_path (pathlib.Path) – Path to the saved model file.

• max_age_hours (int) – Maximum age in hours before model is considered outdated.

• logger (logging.Logger) – Logger object for logging information.

Returns:

True if model is outdated or doesn’t exist, False otherwise.

Return type:

bool

async emhass.command_line.main()

Define the main command line entry function.

This function may take several arguments as inputs. You can type emhass –help to see the list of options:

• action: Set the desired action, options are: perfect-optim, dayahead-optim, naive-mpc-optim, publish-data, forecast-model-fit, forecast-model-predict, forecast-model-tune

• config: Define path to the config.yaml file

• costfun: Define the type of cost function, options are: profit, cost, self-consumption

• log2file: Define if we should log to a file or not

• params: Configuration parameters passed from data/options.json if using the add-on

• runtimeparams: Pass runtime optimization parameters as dictionnary

• debug: Use True for testing purposes

emhass.command_line.main_sync()

Sync wrapper for async main function - used as CLI entry point.

async emhass.command_line.naive_mpc_optim(input_data_dict: dict, logger: Logger, save_data_to_file: bool | None = False, debug: bool | None = False) → DataFrame

Perform a call to the naive Model Predictive Controller optimization routine.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging object) – The passed logger object

• save_data_to_file (bool, optional) – Save optimization results to CSV file

• debug (bool, optional) – A debug option useful for unittests

Returns:

The output data of the optimization

Return type:

pd.DataFrame

async emhass.command_line.perfect_forecast_optim(input_data_dict: dict, logger: Logger, save_data_to_file: bool | None = True, debug: bool | None = False) → DataFrame

Perform a call to the perfect forecast optimization routine.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging object) – The passed logger object

• save_data_to_file (bool, optional) – Save optimization results to CSV file

• debug (bool, optional) – A debug option useful for unittests

Returns:

The output data of the optimization

Return type:

pd.DataFrame

emhass.command_line.prepare_forecast_and_weather_data(input_data_dict: dict, logger: Logger, warn_on_resolution: bool = False) → DataFrame | bool

Prepare forecast data with load costs, production prices, outdoor temperature, and GHI.

This helper function eliminates duplication between dayahead_forecast_optim and naive_mpc_optim.

Parameters:

• input_data_dict (dict) – Dictionary with forecast and input data

• logger (logging.Logger) – Logger object

• warn_on_resolution (bool) – Whether to warn about GHI resolution mismatch

Returns:

Prepared DataFrame or False on error

Return type:

pd.DataFrame | bool

async emhass.command_line.publish_data(input_data_dict: dict, logger: Logger, save_data_to_file: bool | None = False, opt_res_latest: DataFrame | None = None, entity_save: bool | None = False, dont_post: bool | None = False) → DataFrame

Publish the data obtained from the optimization results.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging object) – The passed logger object

• save_data_to_file (bool, optional) – If True we will read data from optimization results in dayahead CSV file

• entity_save (bool, optional) – Save built entities to data_path/entities

• dont_post (bool, optional) – Do not post to Home Assistant. Works with entity_save

Returns:

The output data of the optimization readed from a CSV file in the data folder

Return type:

pd.DataFrame

async emhass.command_line.publish_json(entity: dict, input_data_dict: dict, entity_path: Path, logger: Logger, reference: str | None = '')

Extract saved entity data from .json (in data_path/entities), build entity, post results to post_data

Parameters:

• entity (dict) – json file containing entity data

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• entity_path (Path) – Path for entities folder in data_path

• logger (logging.Logger) – The passed logger object

• reference (str, optional) – String for identifying who ran the function

async emhass.command_line.regressor_model_fit(input_data_dict: dict, logger: Logger, debug: bool | None = False) → MLRegressor

Perform a forecast model fit from training data retrieved from Home Assistant.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging.Logger) – The passed logger object

• debug (Optional[bool], optional) – True to debug, useful for unit testing, defaults to False

async emhass.command_line.regressor_model_predict(input_data_dict: dict, logger: Logger, debug: bool | None = False, mlr: MLRegressor | None = None) → ndarray

Perform a prediction from csv file.

Parameters:

• input_data_dict (dict) – A dictionnary with multiple data used by the action functions

• logger (logging.Logger) – The passed logger object

• debug (Optional[bool], optional) – True to debug, useful for unit testing, defaults to False

async emhass.command_line.retrieve_home_assistant_data(set_type: str, get_data_from_file: bool, retrieve_hass_conf: dict, optim_conf: dict, rh: RetrieveHass, emhass_conf: dict, test_df_literal: str, logger: Logger | None = None) → tuple[bool, DataFrame | None, list | None]

Retrieve data from Home Assistant or file and prepare it for optimization.

async emhass.command_line.set_input_data_dict(emhass_conf: dict, costfun: str, params: str, runtimeparams: str, set_type: str, logger: Logger, get_data_from_file: bool | None = False) → dict

Set up some of the data needed for the different actions.

Parameters:

• emhass_conf (dict) – Dictionary containing the needed emhass paths

• costfun (str) – The type of cost function to use for optimization problem

• params (str) – Configuration parameters passed from data/options.json

• runtimeparams (str) – Runtime optimization parameters passed as a dictionary

• set_type (str) – Set the type of setup based on following type of optimization

• logger (logging object) – The passed logger object

• get_data_from_file (bool, optional) – Use data from saved CSV file (useful for debug)

Returns:

A dictionnary with multiple data used by the action functions

Return type:

dict

async emhass.command_line.weather_forecast_cache(emhass_conf: dict, params: str, runtimeparams: str, logger: Logger) → bool

Perform a call to get forecast function, intend to save results to cache.

Parameters:

• emhass_conf (dict) – Dictionary containing the needed emhass paths

• params (str) – Configuration parameters passed from data/options.json

• runtimeparams (str) – Runtime optimization parameters passed as a dictionary

• logger (logging object) – The passed logger object

Returns:

A bool for function completion

Return type:

bool

emhass.forecast module

class emhass.forecast.Forecast(retrieve_hass_conf: dict, optim_conf: dict, plant_conf: dict, params: str, emhass_conf: dict, logger: Logger, opt_time_delta: int | None = 24, get_data_from_file: bool | None = False)

Bases: object

Generate weather, load and costs forecasts needed as inputs to the optimization.

In EMHASS we have basically 4 forecasts to deal with:

• PV power production forecast (internally based on the weather forecast and the characteristics of your PV plant). This is given in Watts.

• Load power forecast: how much power your house will demand on the next 24h. This is given in Watts.

• PV production selling price forecast: at what price are you selling your excess PV production on the next 24h. This is given in EUR/kWh.

• Load cost forecast: the price of the energy from the grid on the next 24h. This is given in EUR/kWh.

There are methods that are generalized to the 4 forecast needed. For all there forecasts it is possible to pass the data either as a passed list of values or by reading from a CSV file. With these methods it is then possible to use data from external forecast providers.

Then there are the methods that are specific to each type of forecast and that proposed forecast treated and generated internally by this EMHASS forecast class. For the weather forecast a first method (open-meteo) uses a open-meteos API proposing detailed forecasts based on Lat/Lon locations. This method seems stable but as with any scrape method it will fail if any changes are made to the webpage API. Another method (solcast) is using the SolCast PV production forecast service. A final method (solar.forecast) is using another external service: Solar.Forecast, for which just the nominal PV peak installed power should be provided. Search the forecast section on the documentation for examples on how to implement these different methods.

The get_power_from_weather method is proposed here to convert from irradiance data to electrical power. The PVLib module is used to model the PV plant.

The specific methods for the load forecast are a first method (naive) that uses a naive approach, also called persistance. It simply assumes that the forecast for a future period will be equal to the observed values in a past period. The past period is controlled using parameter delta_forecast. A second method (mlforecaster) uses an internal custom forecasting model using machine learning. There is a section in the documentation explaining how to use this method.

Note

This custom machine learning model is introduced from v0.4.0. EMHASS proposed this new mlforecaster class with fit, predict and tune methods. Only the predict method is used here to generate new forecasts, but it is necessary to previously fit a forecaster model and it is a good idea to optimize the model hyperparameters using the tune method. See the dedicated section in the documentation for more help.

For the PV production selling price and Load cost forecasts the privileged method is a direct read from a user provided list of values. The list should be passed as a runtime parameter during the curl to the EMHASS API.

I reading from a CSV file, it should contain no header and the timestamped data should have the following format: 2021-04-29 00:00:00+00:00,287.07 2021-04-29 00:30:00+00:00,274.27 2021-04-29 01:00:00+00:00,243.38 …

The data columns in these files will correspond to the data in the units expected for each forecasting method.

adjust_pv_forecast_data_prep(data: DataFrame) → DataFrame

Prepare data for adjusting the photovoltaic (PV) forecast.

This method aligns the actual PV production data with the forecasted data, adds additional features for analysis, and separates the predictors (X) from the target variable (y).

Parameters:

data (pd.DataFrame) – A DataFrame containing the actual PV production data and the forecasted PV production data.

Returns:

DataFrame with data for adjusted PV model train.

async adjust_pv_forecast_fit(n_splits: int = 5, regression_model: str = 'LassoRegression', debug: bool | None = False) → DataFrame

Fit a regression model to adjust the photovoltaic (PV) forecast.

This method uses historical actual and forecasted PV production data, along with additional solar and date features, to train a regression model. The model is optimized using a grid search with time-series cross-validation.

Parameters:

• n_splits (int, optional) – The number of splits for time-series cross-validation, defaults to 5.

• regression_model (str, optional) – The type of regression model to use. See REGRESSION_METHODS in machine_learning_regressor.py for the authoritative list of supported models. Currently: ‘LinearRegression’, ‘RidgeRegression’, ‘LassoRegression’, ‘ElasticNet’, ‘KNeighborsRegressor’, ‘DecisionTreeRegressor’, ‘SVR’, ‘RandomForestRegressor’, ‘ExtraTreesRegressor’, ‘GradientBoostingRegressor’, ‘AdaBoostRegressor’, ‘MLPRegressor’. Defaults to “LassoRegression”.

• debug (bool, optional) – If True, the model is not saved to disk, useful for debugging, defaults to False.

Returns:

A DataFrame containing the adjusted PV forecast.

Return type:

pd.DataFrame

adjust_pv_forecast_predict(forecasted_pv: DataFrame | None = None) → DataFrame

Predict the adjusted photovoltaic (PV) forecast.

This method uses the trained regression model to predict the adjusted PV forecast based on either the validation data stored in self or a new forecasted PV data passed as input. It applies additional features such as date and solar angles to the forecasted PV production data before making predictions. The solar elevation is used to avoid negative values and to fix values at the beginning and end of the day.

Parameters:

forecasted_pv (pd.DataFrame, optional) – Optional. A DataFrame containing the forecasted PV production data. It must have a DateTime index and a column named “forecast”. If not provided, the method will use self.p_pv_forecast_validation.

Returns:

A DataFrame containing the adjusted PV forecast with additional features.

Return type:

pd.DataFrame

cloud_cover_to_irradiance(cloud_cover: Series, offset: int | None = 35) → DataFrame

Estimates irradiance from cloud cover in the following steps.

1. Determine clear sky GHI using Ineichen model and climatological turbidity.

2. Estimate cloudy sky GHI using a function of cloud_cover

3. Estimate cloudy sky DNI using the DISC model.

4. Calculate DHI from DNI and GHI.

(This function was copied and modified from PVLib)

Parameters:

• cloud_cover (pd.Series) – Cloud cover in %.

• offset (Optional[int], optional) – Determines the minimum GHI., defaults to 35

Returns:

Estimated GHI, DNI, and DHI.

Return type:

pd.DataFrame

static compute_solar_angles(df: DataFrame, latitude: float, longitude: float) → DataFrame

Compute solar angles (elevation, azimuth) based on timestamps and location.

Parameters:

• df – DataFrame with a DateTime index.

• latitude – Latitude of the PV system.

• longitude – Longitude of the PV system.

Returns:

DataFrame with added solar elevation and azimuth.

async get_cached_forecast_data(w_forecast_cache_path) → DataFrame | None

Get cached weather forecast data from file.

Parameters:

w_forecast_cache_path – the path to file.

Returns:

The DataFrame containing the forecasted data, or None when the cache is corrupt/missing or was intentionally deleted (e.g. stale Open-Meteo cache). Callers must handle the None case.

Return type:

pd.DataFrame | None

async get_cached_open_meteo_forecast_json(max_age: int | None = 30, forecast_days: int = 3) → dict

Get weather forecast json from Open-Meteo and cache it for re-use. The response json is cached in the local file system and returned on subsequent calls until it is older than max_age, at which point attempts will be made to replace it with a new version. The cached version will not be overwritten until a new version has been successfully fetched from Open-Meteo. In the event of connectivity issues, the cached version will continue to be returned until such time as a new version can be successfully fetched from Open-Meteo. If you want to force reload, pass max_age value of zero.

Parameters:

• max_age (int, optional) – The maximum age of the cached json file, in minutes, before it is discarded and a new version fetched from Open-Meteo. Defaults to 30 minutes.

• forecast_days (int, optional) – The number of days of forecast data required from Open-Meteo. One additional day is always fetched from Open-Meteo so there is an extra data in the cache. Defaults to 2 days (3 days fetched) to match the prior default.

Returns:

The json containing the Open-Meteo forecast data

Return type:

dict

get_forecast_days_csv(timedelta_days: int | None = 1) → date_range

Get the date range vector of forecast dates that will be used when loading a CSV file.

Returns:

The forecast dates vector

Return type:

pd.date_range

get_forecast_out_from_csv_or_list(df_final: DataFrame, forecast_dates_csv: date_range, csv_path: str, data_list: list | None = None, list_and_perfect: bool | None = False) → DataFrame

Get the forecast data as a DataFrame from a CSV file.

The data contained in the CSV file should be a 24h forecast with the same frequency as the main ‘optimization_time_step’ parameter in the configuration file. The timestamp will not be used and a new DateTimeIndex is generated to fit the timestamp index of the input data in ‘df_final’.

Parameters:

• df_final (pd.DataFrame) – The DataFrame containing the input data.

• forecast_dates_csv (pd.date_range) – The forecast dates vector

• csv_path (str) – The path to the CSV file

Returns:

The data from the CSV file

Return type:

pd.DataFrame

get_load_cost_forecast(df_final: DataFrame, method: str | None = 'hp_hc_periods', csv_path: str | None = 'data_load_cost_forecast.csv', list_and_perfect: bool | None = False) → DataFrame

Get the unit cost for the load consumption based on multiple tariff periods. This is the cost of the energy from the utility in a vector sampled at the fixed freq value.

Parameters:

• df_final (pd.DataFrame) – The DataFrame containing the input data.

• method (str, optional) – The method to be used to generate load cost forecast, the options are ‘hp_hc_periods’ for peak and non-peak hours contractsand ‘csv’ to load a CSV file, defaults to ‘hp_hc_periods’

• csv_path (str, optional) – The path to the CSV file used when method = ‘csv’, defaults to “data_load_cost_forecast.csv”

Returns:

The input DataFrame with one additionnal column appended containing the load cost for each time observation.

Return type:

pd.DataFrame

async get_load_forecast(days_min_load_forecast: int | None = 3, method: str | None = 'typical', csv_path: str | None = 'data_load_forecast.csv', set_mix_forecast: bool | None = False, df_now: DataFrame | None = Empty DataFrame Columns: [] Index: [], use_last_window: bool | None = True, mlf: MLForecaster | None = None, debug: bool | None = False) → Series

Get and generate the load forecast data.

Parameters:

• days_min_load_forecast (int, optional) – The number of last days to retrieve that will be used to generate a naive forecast, defaults to 3

• method (str, optional) – The method to be used to generate load forecast, the options are ‘typical’ for a typical household load consumption curve, are ‘naive’ for a persistence model, ‘mlforecaster’ for using a custom previously fitted machine learning model, ‘csv’ to read the forecast from a CSV file and ‘list’ to use data directly passed at runtime as a list of values. Defaults to ‘typical’.

• csv_path (str, optional) – The path to the CSV file used when method = ‘csv’, defaults to “/data/data_load_forecast.csv”

• set_mix_forecast (Bool, optional) – Use a mixed forecast strategy to integrate now/current values.

• df_now (pd.DataFrame, optional) – The DataFrame containing the now/current data.

• use_last_window (Bool, optional) – True if the ‘last_window’ option should be used for the custom machine learning forecast model. The ‘last_window=True’ means that the data that will be used to generate the new forecast will be freshly retrieved from Home Assistant. This data is needed because the forecast model is an auto-regressive model with lags. If ‘False’ then the data using during the model train is used.

• mlf (mlforecaster, optional) – The ‘mlforecaster’ object previously trained. This is mainly used for debug and unit testing. In production the actual model will be read from a saved pickle file.

• debug (Bool, optional) – The DataFrame containing the now/current data.

Returns:

The DataFrame containing the electrical load power in Watts

Return type:

pd.DataFrame

static get_mix_forecast(df_now: DataFrame, df_forecast: DataFrame, alpha: float, beta: float, col: str, ignore_pv_feedback: bool = False) → DataFrame

A simple correction method for forecasted data using the current real values of a variable.

Parameters:

• df_now (pd.DataFrame) – The DataFrame containing the current/real values

• df_forecast (pd.DataFrame) – The DataFrame containing the forecast data

• alpha (float) – A weight for the forecast data side

• beta (float) – A weight for the current/real values sied

• col (str) – The column variable name

• ignore_pv_feedback (bool) – If True, bypass mixing and return original forecast (used during curtailment)

Returns:

The output DataFrame with the corrected values

Return type:

pd.DataFrame

get_power_from_weather(df_weather: DataFrame, set_mix_forecast: bool | None = False, df_now: DataFrame | None = Empty DataFrame Columns: [] Index: []) → Series

Convert weather forecast data into electrical power.

Parameters:

• df_weather (pd.DataFrame) – The DataFrame containing the weather forecasted data. This DF should be generated by the ‘get_weather_forecast’ method or at least contain the same columns names filled with proper data.

• set_mix_forecast (Bool, optional) – Use a mixed forecast strategy to integrate now/current values.

• df_now (pd.DataFrame) – The DataFrame containing the now/current data.

Returns:

The DataFrame containing the electrical power in Watts

Return type:

pd.DataFrame

get_prod_price_forecast(df_final: DataFrame, method: str | None = 'constant', csv_path: str | None = 'data_prod_price_forecast.csv', list_and_perfect: bool | None = False) → DataFrame

Get the unit power production price for the energy injected to the grid.This is the price of the energy injected to the utility in a vector sampled at the fixed freq value.

Parameters:

• df_input_data (pd.DataFrame) – The DataFrame containing all the input data retrieved from hass

• method (str, optional) – The method to be used to generate the production price forecast, the options are ‘constant’ for a fixed constant value and ‘csv’to load a CSV file, defaults to ‘constant’

• csv_path (str, optional) – The path to the CSV file used when method = ‘csv’, defaults to “/data/data_load_cost_forecast.csv”

Returns:

The input DataFrame with one additionnal column appended containing the power production price for each time observation.

Return type:

pd.DataFrame

static get_typical_load_forecast(data, forecast_date)

Forecast the load profile for the next day based on historic data.

Parameters:

• data (pd.DataFrame) – A DataFrame with a DateTimeIndex containing the historic load data. Must include a ‘load’ column.

• forecast_date (pd.Timestamp) – The date for which the forecast will be generated.

Returns:

A Series with the forecasted load profile for the next day and a list of days used to calculate the forecast.

Return type:

tuple (pd.Series, list)

async get_weather_forecast(method: str | None = 'open-meteo', csv_path: str | None = 'data_weather_forecast.csv', use_legacy_pvlib: bool | None = False) → DataFrame

Get and generate weather forecast data.

Parameters:

method (str, optional) – The desired method, options are ‘open-meteo’, ‘csv’, ‘list’, ‘solcast’ and ‘solar.forecast’. Defaults to ‘open-meteo’.

Returns:

The DataFrame containing the forecasted data

Return type:

pd.DataFrame

static resample_data(data, freq, current_freq)

Resample a DataFrame with a custom frequency.

Parameters:

• data (pd.DataFrame) – Original time series data with a DateTimeIndex.

• freq (pd.Timedelta) – Desired frequency for resampling (e.g., pd.Timedelta(“10min”)).

Returns:

Resampled data at the specified frequency.

Return type:

pd.DataFrame

async set_cached_forecast_data(w_forecast_cache_path, data) → DataFrame

Set generated weather forecast data to file. Trim data to match the original requested forecast dates

Parameters:

w_forecast_cache_path – the path to file.

Param:

The DataFrame containing the forecasted data

Type:

pd.DataFrame

Returns:

The DataFrame containing the forecasted data

Return type:

pd.DataFrame

emhass.machine_learning_forecaster module

class emhass.machine_learning_forecaster.MLForecaster(data: DataFrame, model_type: str, var_model: str, sklearn_model: str, num_lags: int, emhass_conf: dict, logger: Logger)

Bases: object

A forecaster class using machine learning models with auto-regressive approach and featuresbased on timestamp information (hour, day, week, etc).

This class uses the skforecast module and the machine learning models are from scikit-learn.

It exposes three main methods:

• fit: to train a model with the passed data.

• predict: to obtain a forecast from a pre-trained model.

• tune: to optimize the models hyperparameters using bayesian optimization.

async fit(split_date_delta: str | None = '48h', perform_backtest: bool | None = False) → tuple[DataFrame, DataFrame]

The fit method to train the ML model.

Parameters:

• split_date_delta (Optional[str], optional) – The delta from now to split_date_delta that will be used as the test period to evaluate the model, defaults to ‘48h’

• perform_backtest (Optional[bool], optional) – If True then a back testing routine is performed to evaluate the performance of the model on the complete train set, defaults to False

Returns:

The DataFrame containing the forecast data results without and with backtest

Return type:

Tuple[pd.DataFrame, pd.DataFrame]

async static generate_exog(data_last_window, periods, var_name)

Generate the exogenous data for future timestamps.

static get_lags_list_from_frequency(freq: Timedelta) → list[int]

Calculate appropriate lag values based on data frequency.

The lags represent different time horizons (6h, 12h, 1d, 1.5d, 2d, 2.5d, 3d). This method scales these horizons according to the actual data frequency.

Parameters:

freq (pd.Timedelta) – The frequency of the data as a pandas Timedelta

Returns:

A list of lag values appropriate for the data frequency

Return type:

list[int]

async static interpolate_async(data: DataFrame) → DataFrame

Interpolate missing values asynchronously.

static neg_r2_score(y_true, y_pred)

The negative of the r2 score.

async predict(data_last_window: DataFrame | None = None) → Series

The predict method to generate forecasts from a previously fitted ML model.

Parameters:

data_last_window (Optional[pd.DataFrame], optional) – The data that will be used to generate the new forecast, this will be freshly retrieved from Home Assistant. This data is needed because the forecast model is an auto-regressive model with lags. If not passed then the data used during the model train is used, defaults to None

Returns:

A pandas series containing the generated forecasts.

Return type:

pd.Series

async tune(split_date_delta: str | None = '48h', n_trials: int = 10, debug: bool | None = False) → DataFrame

Tuning a previously fitted model using bayesian optimization.

Parameters:

• split_date_delta (Optional[str], optional) – The delta from now to split_date_delta that will be used as the test period to evaluate the model, defaults to ‘48h’. This define the training/validation split for the tuning process.

• debug (Optional[bool], optional) – Set to True for testing and faster optimizations, defaults to False

• n_trials (Optional[int], optional) – Number of trials for bayesian optimization, defaults to 10

Returns:

The DataFrame with the forecasts using the optimized model.

Return type:

pd.DataFrame

emhass.optimization module

class emhass.optimization.Optimization(retrieve_hass_conf: dict, optim_conf: dict, plant_conf: dict, var_load_cost: str, var_prod_price: str, costfun: str, emhass_conf: dict, logger: Logger, opt_time_delta: int | None = 24, num_timesteps: int | None = None)

Bases: object

Optimize the deferrable load and battery energy dispatch problem using the linear programming optimization technique. All equipement equations, including the battery equations are hence transformed in a linear form.

This class methods are:

• perform_optimization

• perform_perfect_forecast_optim

• perform_dayahead_forecast_optim

• perform_naive_mpc_optim

perform_dayahead_forecast_optim(df_input_data: DataFrame, p_pv: Series, p_load: Series, stage_times: dict[str, float] | None = None) → DataFrame

Perform a day-ahead optimization task using real forecast data. This type of optimization is intented to be launched once a day.

Parameters:

• df_input_data (pandas.DataFrame) – A DataFrame containing all the input data used for the optimization, notably the unit load cost for power consumption.

• p_pv (pandas.DataFrame) – The forecasted PV power production.

• p_load (pandas.DataFrame) – The forecasted Load power consumption. This power should not include the power from the deferrable load that we want to find.

• stage_times (dict, optional) – Optional dict to record nested sub-stage timings (optim_solve.build / optim_solve.solve / optim_solve.extract).

Returns:

opt_res: A DataFrame containing the optimization results

Return type:

pandas.DataFrame

perform_naive_mpc_optim(df_input_data: DataFrame, p_pv: Series, p_load: Series, prediction_horizon: int, soc_init: float | None = None, soc_final: float | None = None, def_total_hours: list | None = None, def_total_timestep: list | None = None, def_start_timestep: list | None = None, def_end_timestep: list | None = None, stage_times: dict[str, float] | None = None) → DataFrame

Perform a naive approach to a Model Predictive Control (MPC). This implementaion is naive because we are not using the formal formulation of a MPC. Only the sense of a receiding horizon is considered here. This optimization is more suitable for higher optimization frequency, ex: 5min.

Parameters:

• df_input_data (pandas.DataFrame) – A DataFrame containing all the input data used for the optimization, notably the unit load cost for power consumption.

• p_pv (pandas.DataFrame) – The forecasted PV power production.

• p_load (pandas.DataFrame) – The forecasted Load power consumption. This power should not include the power from the deferrable load that we want to find.

• prediction_horizon (int) – The prediction horizon of the MPC controller in number of optimization time steps.

• soc_init (float) – The initial battery SOC for the optimization. This parameter is optional, if not given soc_init = soc_final = soc_target from the configuration file.

• soc_final – The final battery SOC for the optimization. This parameter is optional, if not given soc_init = soc_final = soc_target from the configuration file.

• def_total_timestep (list) – The functioning timesteps for this iteration for each deferrable load. (For continuous deferrable loads: functioning timesteps at nominal power)

• def_total_hours (list) – The functioning hours for this iteration for each deferrable load. (For continuous deferrable loads: functioning hours at nominal power)

• def_start_timestep (list) – The timestep as from which each deferrable load is allowed to operate.

• def_end_timestep (list) – The timestep before which each deferrable load should operate.

Returns:

opt_res: A DataFrame containing the optimization results

Return type:

pandas.DataFrame

perform_optimization(data_opt: DataFrame, p_pv: array, p_load: array, unit_load_cost: array, unit_prod_price: array, soc_init: float | None = None, soc_final: float | None = None, def_total_hours: list | None = None, def_total_timestep: list | None = None, def_start_timestep: list | None = None, def_end_timestep: list | None = None, def_init_temp: list | None = None, min_power_of_deferrable_loads: list | None = None, debug: bool | None = False, stage_times: dict[str, float] | None = None) → DataFrame

Perform the actual optimization using Convex Programming (CVXPY). Includes automatic fallback to relaxed LP if MILP fails or times out.

If stage_times is provided, the wall-clock duration of three internal phases is recorded under the keys optim_solve.build, optim_solve.solve and optim_solve.extract. These nest under the existing optim_solve parent timer in command_line.py and sum to it within a few milliseconds.

perform_perfect_forecast_optim(df_input_data: DataFrame, days_list: date_range) → DataFrame

Perform an optimization on historical data (perfectly known PV production).

Parameters:

• df_input_data (pandas.DataFrame) – A DataFrame containing all the input data used for the optimization, notably photovoltaics and load consumption powers.

• days_list (list) – A list of the days of data that will be retrieved from hass and used for the optimization task. We will retrieve data from now and up to days_to_retrieve days

Returns:

opt_res: A DataFrame containing the optimization results

Return type:

pandas.DataFrame

update_battery_power_limits(plant_conf: dict) → None

Update battery charge/discharge power-limit Parameters from plant_conf.

Called on cache hit to sync runtime power-limit values without rebuilding constraints. Mirrors update_thermal_start_temps.

Parameters:

plant_conf – The plant configuration containing battery_charge_power_max / battery_discharge_power_max

update_thermal_params(optim_conf: dict, data_opt: DataFrame, p_load: ndarray) → None

Update all thermal parameters from optim_conf and data_opt.

Called on cache hit to sync all runtime thermal parameters without rebuilding constraints. This includes start_temperature, outdoor_temp forecasts, min/max temps, and derived values like thermal_losses, heating_demand, and heatpump_cops.

Parameters:

• optim_conf – The optimization configuration containing def_load_config

• data_opt – DataFrame with forecast data (outdoor_temperature_forecast, ghi, etc.)

• p_load – Load power forecast array (for internal gains calculation)

update_thermal_start_temps(optim_conf: dict) → None

Update thermal start temperature parameters from optim_conf.

Called on cache hit to sync runtime thermal parameters without rebuilding constraints. This is a convenience wrapper that only updates start_temp. For full updates including forecasts, use update_thermal_params().

Parameters:

optim_conf – The optimization configuration containing def_load_config

static validate_def_timewindow(start: int, end: int, min_steps: int, window: int) → tuple[int, int, str]

Helper function to validate (and if necessary: correct) the defined optimization window of a deferrable load.

Parameters:

• start (int) – Start timestep of the optimization window of the deferrable load

• end (int) – End timestep of the optimization window of the deferrable load

• min_steps (int) – Minimal timesteps during which the load should operate (at nominal power)

• window (int) – Total number of timesteps in the optimization window

Returns:

start_validated: Validated start timestep of the optimization window of the deferrable load

Return type:

int

Returns:

end_validated: Validated end timestep of the optimization window of the deferrable load

Return type:

int

Returns:

warning: Any warning information to be returned from the validation steps

Return type:

string

emhass.retrieve_hass module

class emhass.retrieve_hass.RetrieveHass(hass_url: str, long_lived_token: str, freq: Timedelta, time_zone: timezone, params: str, emhass_conf: dict, logger: Logger, get_data_from_file: bool | None = False)

Bases: object

Retrieve data from Home Assistant using the restful API.

This class allows the user to retrieve data from a Home Assistant instance using the provided restful API (https://developers.home-assistant.io/docs/api/rest/)

This class methods are:

• get_data: to retrieve the actual data from hass

• prepare_data: to apply some data treatment in preparation for the optimization task

• post_data: Post passed data to hass

async close() → None

Close the persistent HTTP session.

Should be called when the RetrieveHass instance is no longer needed to properly release resources.

static get_attr_data_dict(data_df: DataFrame, idx: int, entity_id: str, device_class: str, unit_of_measurement: str, friendly_name: str, list_name: str, state: float, decimals: int = 2) → dict

async get_data(days_list: date_range, var_list: list, minimal_response: bool | None = False, significant_changes_only: bool | None = False, test_url: str | None = 'empty') → None

Retrieve the actual data from hass.

Parameters:

• days_list (pandas.date_range) – A list of days to retrieve. The ISO format should be used and the timezone is UTC. The frequency of the data_range should be freq=’D’

• var_list (list) – The list of variables to retrive from hass. These should be the exact name of the sensor in Home Assistant. For example: [‘sensor.home_load’, ‘sensor.home_pv’]

• minimal_response (bool, optional) – Retrieve a minimal response using the hass restful API, defaults to False

• significant_changes_only (bool, optional) – Retrieve significant changes only using the hass restful API, defaults to False

Returns:

The DataFrame populated with the retrieved data from hass

Return type:

pandas.DataFrame

get_data_influxdb(days_list: 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.

Parameters:

• days_list (pandas.date_range) – A list of days to retrieve data for

• var_list (list) – List of sensor entity IDs to retrieve

Returns:

Success status of data retrieval

Return type:

bool

async get_data_websocket(days_list: date_range, var_list: list[str]) → bool

Retrieve the actual data from hass.

Parameters:

• days_list (pandas.date_range) – A list of days to retrieve. The ISO format should be used and the timezone is UTC. The frequency of the data_range should be freq=’D’

• var_list (list) – The list of variables to retrive from hass. These should be the exact name of the sensor in Home Assistant. For example: [‘sensor.home_load’, ‘sensor.home_pv’]

Returns:

The DataFrame populated with the retrieved data from hass

Return type:

pandas.DataFrame

async get_ha_config()

Extract some configuration data from HA.

Return type:

bool

async get_ha_config_websocket() → dict[str, Any]

Get Home Assistant configuration.

async post_data(data_df: DataFrame, idx: int, entity_id: str, device_class: str, unit_of_measurement: str, friendly_name: str, type_var: str, publish_prefix: str | None = '', save_entities: bool | None = False, logger_levels: str | None = 'info', dont_post: bool | None = False) → None

Post passed data to hass using REST API.

Note

This method ALWAYS uses the REST API for posting data to Home Assistant, regardless of the use_websocket setting. WebSocket is only used for data retrieval, not for publishing/posting data.

Parameters:

• data_df (pd.DataFrame) – The DataFrame containing the data that will be posted to hass. This should be a one columns DF or a series.

• idx (int) – The int index of the location of the data within the passed DataFrame. We will post just one value at a time.

• entity_id (str) – The unique entity_id of the sensor in hass.

• device_class (str) – The HASS device class for the sensor.

• unit_of_measurement (str) – The units of the sensor.

• friendly_name (str) – The friendly name that will be used in the hass frontend.

• type_var (str) – A variable to indicate the type of variable: power, SOC, etc.

• publish_prefix (str, optional) – A common prefix for all published data entity_id.

• save_entities (bool, optional) – if entity data should be saved in data_path/entities

• logger_levels (str, optional) – set logger level, info or debug, to output

• dont_post (bool, optional) – dont post to HA

prepare_data(var_load: str, load_negative: bool, set_zero_min: bool, var_replace_zero: list[str], var_interp: list[str], skip_renaming: bool = False) → bool

Apply some data treatment in preparation for the optimization task.

Parameters:

• var_load (str) – The name of the variable for the household load consumption.

• load_negative (bool, optional) – Set to True if the retrived load variable is negative by convention, defaults to False

• set_zero_min (bool, optional) – A special treatment for a minimum value saturation to zero. Values below zero are replaced by nans, defaults to True

• var_replace_zero (list, optional) – A list of retrived variables that we would want to replace nans with zeros, defaults to None

• var_interp (list, optional) – A list of retrived variables that we would want to interpolate nan values using linear interpolation, defaults to None

Returns:

The DataFrame populated with the retrieved data from hass and after the data treatment

Return type:

pandas.DataFrame

emhass.utils module

emhass.utils.add_date_features(data: DataFrame, timestamp: str | None = None, date_features: list[str] | None = None) → DataFrame

Add date-related features from a DateTimeIndex or a timestamp column.

Parameters:

• data (pd.DataFrame) – The input DataFrame.

• timestamp (Optional[str]) – The column containing the timestamp (optional if DataFrame has a DateTimeIndex).

• date_features (Optional[List[str]]) – List of date features to extract (default: all).

Returns:

The DataFrame with added date features.

Return type:

pd.DataFrame

emhass.utils.apply_heating_curve(heating_curve: dict, outdoor_temperature_forecast: ndarray | Series) → ndarray

Compute per-slot supply temperature from a weather-compensated heating curve.

A heating curve specifies how the heat source modulates its supply temperature in response to outdoor temperature, the way every modern boiler / heat pump controller does. The linear form supported here:

T_supply(t) = clip(offset - slope * T_outdoor(t), min_supply, max_supply)

Parameters:

• heating_curve – dict with required slope and offset (both °C), plus optional min_supply (default 25°C) and max_supply (default 70°C).

• outdoor_temperature_forecast – Per-slot outdoor temperature in °C.

Returns:

Per-slot supply temperature array in °C.

async emhass.utils.build_config(emhass_conf: dict, logger: Logger, defaults_path: str, config_path: str | None = None, legacy_config_path: str | None = None) → dict

Retrieve parameters from configuration files. priority order (low - high) = defaults_path, config_path legacy_config_path

Parameters:

• emhass_conf (dict) – Dictionary containing the needed emhass paths

• logger (logging.Logger) – The logger object

• defaults_path (str) – path to config file for parameter defaults (config_defaults.json)

• config_path (str) – path to the main configuration file (config.json)

• legacy_config_path (str) – path to legacy config file (config_emhass.yaml)

Returns:

The built config dictionary

Return type:

dict

async emhass.utils.build_legacy_config_params(emhass_conf: dict[str, Path], legacy_config: dict[str, str], logger: Logger) → dict[str, str]

Build a config dictionary with legacy config_emhass.yaml file. Uses the associations file to convert parameter naming conventions (to config.json/config_defaults.json). Extracts the parameter values and formats to match config.json.

Parameters:

• emhass_conf (dict) – Dictionary containing the needed emhass paths

• legacy_config (dict) – The legacy config dictionary

• logger (logging.Logger) – The logger object

Returns:

The built config dictionary

Return type:

dict

async emhass.utils.build_params(emhass_conf: dict[str, Path], params_secrets: dict[str, str | float], config: dict[str, str], logger: Logger) → dict[str, dict]

Build the main params dictionary from the config and secrets Appends configuration catagories used by emhass to the parameters. (with use of the associations file as a reference)

Parameters:

• emhass_conf (dict[str, pathlib.Path]) – Dictionary containing the needed emhass paths

• params_secrets (dict[str, str | float]) – The dictionary containing the built secret variables

• config (dict[str, str]) – The dictionary of built config parameters

• logger (logging.Logger) – The logger object

Returns:

The built param dictionary

Return type:

dict[str, dict]

async emhass.utils.build_secrets(emhass_conf: dict[str, Path], logger: Logger, argument: dict[str, str] | None = None, options_path: str | None = None, secrets_path: str | None = None, no_response: bool = False) → tuple[dict[str, Path], dict[str, str | float]]

Retrieve and build parameters from secrets locations (ENV, ARG, Secrets file (secrets_emhass.yaml/options.json) and/or Home Assistant (via API)) priority order (lwo to high) = Defaults (written in function), ENV, Options json file, Home Assistant API, Secrets yaml file, Arguments

Parameters:

• emhass_conf (dict) – Dictionary containing the needed emhass paths

• logger (logging.Logger) – The logger object

• argument (dict) – dictionary of secrets arguments passed (url,key)

• options_path (str) – path to the options file (options.json) (usually provided by EMHASS-Add-on)

• secrets_path (str) – path to secrets file (secrets_emhass.yaml)

• no_response (bool) – bypass get request to Home Assistant (json response errors)

Returns:

Updated emhass_conf, the built secrets dictionary

Return type:

Tuple[dict, dict]:

emhass.utils.calculate_cop_heatpump(supply_temperature: float | ndarray | Series, carnot_efficiency: float, outdoor_temperature_forecast: ndarray | Series, mode: str = 'heat') → ndarray

Calculate heat pump Coefficient of Performance (COP) for each timestep in the prediction horizon.

The COP is calculated using a Carnot-based formula. Two modes are supported:

• mode='heat' (default):

  \[COP_{heat}(h) = \eta_{carnot} \times \frac{T_{supply\_K}}{T_{supply\_K} - T_{outdoor\_K}(h)}\]

• mode='cool':

  \[COP_{cool}(h) = \eta_{carnot} \times \frac{T_{supply\_K}}{T_{outdoor\_K}(h) - T_{supply\_K}}\]

Where temperatures are converted to Kelvin (K = °C + 273.15).

This formula models real heat pump behavior where COP decreases as the temperature lift (difference between supply and outdoor temperature) increases. The carnot_efficiency factor represents the real-world efficiency as a fraction of the ideal Carnot cycle efficiency.

Parameters:

• supply_temperature (float or np.ndarray or pd.Series) – The heat pump supply temperature in degrees Celsius. Can be a scalar (constant supply T) or an array/Series matching the length of outdoor_temperature_forecast for weather-compensated supply T (heating curve). Typical scalar values: 30-40°C for underfloor heating, 50-70°C for radiator systems.

• carnot_efficiency (float) – Real-world efficiency factor as fraction of ideal Carnot cycle. Typical range: 0.35-0.50 (35-50%). Default in thermal battery config: 0.4 (40%). Higher values represent more efficient heat pumps.

• outdoor_temperature_forecast (np.ndarray or pd.Series) – Array of outdoor temperature forecasts in degrees Celsius, one value per timestep in the prediction horizon.

• mode (str) – Operating mode, either "heat" or "cool". In cooling mode, the Carnot lift uses T_outdoor - T_supply so warm weather no longer collapses to COP=1.0 as in heating-only validation.

Returns:

Array of COP values for each timestep, same length as outdoor_temperature_forecast. Typical COP range: 2-6 for normal operating conditions.

Return type:

np.ndarray

Example:

>>> supply_temp = 35.0  # °C, underfloor heating
>>> carnot_eff = 0.4  # 40% of ideal Carnot efficiency
>>> outdoor_temps = np.array([0.0, 5.0, 10.0, 15.0, 20.0])
>>> cops = calculate_cop_heatpump(supply_temp, carnot_eff, outdoor_temps)
>>> cops
array([3.521..., 4.108..., 4.926..., 6.163..., 8.217...])
>>> # At 5°C outdoor: COP = 0.4 × 308.15K / 30K = 4.11

emhass.utils.calculate_heating_demand(specific_heating_demand: float, floor_area: float, outdoor_temperature_forecast: ndarray | Series, base_temperature: float = 18.0, annual_reference_hdd: float = 3000.0, optimization_time_step: int | None = None) → ndarray

Calculate heating demand per timestep based on heating degree days method.

Uses heating degree days (HDD) to calculate heating demand based on outdoor temperature forecast, specific heating demand, and floor area. The specific heating demand should be calibrated to the annual reference HDD value.

Parameters:

• specific_heating_demand (float) – Specific heating demand in kWh/m²/year (calibrated to annual_reference_hdd)

• floor_area (float) – Floor area in m²

• outdoor_temperature_forecast (np.ndarray | pd.Series) – Outdoor temperature forecast in °C for each timestep

• base_temperature (float, optional) – Base temperature for HDD calculation in °C, defaults to 18.0 (European standard)

• annual_reference_hdd (float, optional) – Annual reference HDD value for normalization, defaults to 3000.0 (Central Europe)

• optimization_time_step (int | None, optional) – Optimization time step in minutes. If None, automatically infers from pandas Series DatetimeIndex frequency. Falls back to 30 minutes if not inferrable.

Returns:

Array of heating demand values (kWh) per timestep

Return type:

np.ndarray

emhass.utils.calculate_heating_demand_physics(u_value: float, envelope_area: float, ventilation_rate: float, heated_volume: float, indoor_target_temperature: float, outdoor_temperature_forecast: ndarray | Series, optimization_time_step: int, solar_irradiance_forecast: ndarray | Series | None = None, window_area: float | None = None, shgc: float = 0.6, internal_gains_forecast: ndarray | Series | None = None, internal_gains_factor: float = 0.0) → ndarray

Calculate heating demand per timestep based on building physics heat loss model.

More accurate than HDD method as it directly calculates transmission and ventilation losses based on building thermal properties. Optionally accounts for solar gains through windows to reduce heating demand.

Parameters:

• u_value (float) – Overall thermal transmittance (U-value) in W/(m²·K). Typical values: - 0.2-0.3: Well-insulated modern building - 0.4-0.6: Average insulation - 0.8-1.2: Poor insulation / old building

• envelope_area (float) – Total building envelope area (walls + roof + floor + windows) in m²

• ventilation_rate (float) – Air changes per hour (ACH). Typical values: - 0.3-0.5: Well-sealed modern building with controlled ventilation - 0.5-1.0: Average building - 1.0-2.0: Leaky old building

• heated_volume (float) – Total heated volume in m³

• indoor_target_temperature (float) – Target indoor temperature in °C

• outdoor_temperature_forecast (np.ndarray | pd.Series) – Outdoor temperature forecast in °C for each timestep

• optimization_time_step (int) – Optimization time step in minutes

• solar_irradiance_forecast (np.ndarray | pd.Series | None, optional) – Global Horizontal Irradiance (GHI) in W/m² for each timestep. If provided along with window_area, solar gains will be subtracted from heating demand.

• window_area (float | None, optional) – Total window area in m². If provided along with solar_irradiance_forecast, solar gains will reduce heating demand. Typical values: 15-25% of floor area.

• shgc (float, optional) – Solar Heat Gain Coefficient (dimensionless, 0-1). Fraction of solar radiation that becomes heat inside the building. Typical values: - 0.5-0.6: Modern low-e double-glazed windows - 0.6-0.7: Standard double-glazed windows - 0.7-0.8: Single-glazed windows Default: 0.6

• internal_gains_forecast (np.ndarray | pd.Series | None, optional) – Electrical load power forecast in W for each timestep. If provided along with internal_gains_factor > 0, internal gains from electrical appliances will be subtracted from heating demand.

• internal_gains_factor (float, optional) – Factor (0-1) representing what fraction of electrical load becomes useful internal heat gains. Typical values: - 0.0: No internal gains considered (default, backwards compatible) - 0.5-0.7: Conservative estimate (some heat lost to ventilation/drains) - 0.8-0.9: Most electrical energy becomes heat (well-insulated building) - 1.0: All electrical energy becomes internal heat (theoretical maximum) Default: 0.0

Returns:

Array of heating demand values (kWh) per timestep

Return type:

np.ndarray

Example:

>>> outdoor_temps = np.array([5, 8, 12, 15])
>>> ghi = np.array([0, 100, 400, 600])  # W/m²
>>> demand = calculate_heating_demand_physics(
...     u_value=0.3,
...     envelope_area=400,
...     ventilation_rate=0.5,
...     heated_volume=250,
...     indoor_target_temperature=20,
...     outdoor_temperature_forecast=outdoor_temps,
...     optimization_time_step=30,
...     solar_irradiance_forecast=ghi,
...     window_area=50,
...     shgc=0.6
... )

emhass.utils.calculate_surface_solar_gain(hc: dict, ghi_forecast: ndarray | None, optimization_time_step_minutes: float, length: int | None = None) → ndarray | None

Compute per-timestep solar energy absorbed by an exposed thermal mass surface.

Intended for thermal_battery configs that model a thermal store exposed directly to sunlight (a pool, an outdoor tank, a solar-thermal collector routed into a buffer). The gain is independent of the heater and acts as a negative term on heating_demand (i.e. the optimizer needs less pumped heat to maintain temperature when there is solar gain).

Reuses the existing GHI forecast that EMHASS already fetches for PV. No second weather API call is required.

Parameters:

• hc –

  The thermal_battery sub-config dict. Reads two keys: - solar_absorption_area: effective horizontal absorption surface (m²). - solar_absorption_factor: fraction of GHI absorbed by the thermal

  mass (typical pool with no cover: 0.7-0.9; covered pool: 0.2-0.4). Defaults to 0.7 if absent.

• ghi_forecast – Global horizontal irradiance forecast in W/m² per timestep. Pass None or zero-length array to skip gain.

• optimization_time_step_minutes – Timestep duration in minutes.

• length – Truncate / pad the returned array to this length.

Returns:

Solar gain in kWh per timestep, or None if not applicable.

emhass.utils.calculate_thermal_loss_signed(outdoor_temperature_forecast: ndarray | Series, indoor_temperature: float, base_loss: float) → ndarray

Calculate signed thermal loss factor based on indoor/outdoor temperature difference.

SIGN CONVENTION: - Positive (+loss): outdoor < indoor → heat loss, building cools, heating required - Negative (-loss): outdoor ≥ indoor → heat gain, building warms passively

Formula: loss * (1 - 2 * Hot(h)), where Hot(h) = 1 if outdoor ≥ indoor, else 0. Based on Langer & Volling (2020) Equation B.13.

Parameters:

• outdoor_temperature_forecast (np.ndarray or pd.Series) – Outdoor temperature forecast (°C)

• indoor_temperature (float) – Indoor/target temperature threshold (°C)

• base_loss (float) – Base thermal loss coefficient in kW

Returns:

Signed loss array (positive = heat loss, negative = heat gain)

Return type:

np.ndarray

emhass.utils.check_def_loads(num_def_loads: int, parameter: list[dict], default: str | float, parameter_name: str, logger: Logger) → list[dict]

Check parameter lists with deferrable loads number, if they do not match, enlarge to fit.

Parameters:

• num_def_loads (int) – Total number deferrable loads

• parameter (list[dict]) – parameter config dict containing paramater

• default (str | int | float) – default value for parameter to pad missing

• parameter_name (str) – name of parameter

• logger (logging.Logger) – The logger object

Returns:

parameter list

Return type:

list[dict]

emhass.utils.clean_sensor_column_names(df: DataFrame, timestamp_col: str) → DataFrame

Clean sensor column names by removing ‘sensor.’ prefix.

Parameters:

• df (pd.DataFrame) – Input DataFrame with sensor columns

• timestamp_col (str) – Name of timestamp column to preserve

Returns:

DataFrame with cleaned column names

Return type:

pd.DataFrame

emhass.utils.compile_heat_topology(topology: dict) → dict

Compile a heat-topology graph descriptor into flat optim_conf fields.

The graph model lets users declare a small directed graph of sources, storage, consumers, flows, and actuator_groups. This function translates that high-level descriptor into the primitives the optimizer already understands:

• Each flow (source, storage) becomes one deferrable load with a thermal_source block in def_load_config.

• Each storage becomes one entry in shared_thermal_tanks with load_ids pointing at the deferrable loads feeding it.

• Each consumer is folded into its target storage (its profile or building model populates the storage’s draw_off_demand / u_value etc.).

• Each actuator_group becomes one entry in deferrable_load_groups.

• Each source’s cost_track reference resolves to an entry in cost_forecast_per_deferrable_load.

Returns a dict of optim_conf fields to merge into the live optim_conf:

number_of_deferrable_loads, nominal_power_of_deferrable_loads, minimum_power_of_deferrable_loads, treat_deferrable_load_as_semi_cont, operating_hours_of_each_deferrable_load, def_load_config, shared_thermal_tanks, deferrable_load_groups, cost_forecast_per_deferrable_load.

Validation: missing source/storage references, duplicated ids, and consumers targeting unknown storage all raise ValueError with the offending field path.

emhass.utils.get_days_list(days_to_retrieve: int) → DatetimeIndex

Get list of past days from today to days_to_retrieve.

Parameters:

days_to_retrieve (int) – Total number of days to retrieve from the past

Returns:

The list of days

Return type:

pd.DatetimeIndex

emhass.utils.get_forecast_dates(freq: int, delta_forecast: int, time_zone: tzinfo, timedelta_days: int | None = 0) → DatetimeIndex

Get the date_range list of the needed future dates using the delta_forecast parameter.

Parameters:

• freq (int) – Optimization time step.

• delta_forecast (int) – Number of days to forecast in the future to be used for the optimization.

• timedelta_days (Optional[int], optional) – Number of truncated days needed for each optimization iteration, defaults to 0

Returns:

A list of future forecast dates.

Return type:

pd.core.indexes.datetimes.DatetimeIndex

emhass.utils.get_injection_dict(df: DataFrame, plot_size: int | None = 1366) → dict

Build a dictionary with graphs and tables for the webui.

Parameters:

• df (pd.DataFrame) – The optimization result DataFrame

• plot_size (Optional[int], optional) – Size of the plot figure in pixels, defaults to 1366

Returns:

A dictionary containing the graphs and tables in html format

Return type:

dict

emhass.utils.get_injection_dict_forecast_model_fit(df_fit_pred: pd.DataFrame, mlf: MLForecaster) → dict

Build a dictionary with graphs and tables for the webui for special MLF fit case.

Parameters:

• df_fit_pred (pd.DataFrame) – The fit result DataFrame

• mlf (MLForecaster) – The MLForecaster object

Returns:

A dictionary containing the graphs and tables in html format

Return type:

dict

emhass.utils.get_injection_dict_forecast_model_tune(df_pred_optim: pd.DataFrame, mlf: MLForecaster) → dict

Build a dictionary with graphs and tables for the webui for special MLF tune case.

Parameters:

• df_pred_optim (pd.DataFrame) – The tune result DataFrame

• mlf (MLForecaster) – The MLForecaster object

Returns:

A dictionary containing the graphs and tables in html format

Return type:

dict

emhass.utils.get_keys_to_mask() → list[str]

Return a list of sensitive configuration keys that should be masked in logs or treated specially in the UI (e.g., secrets).

emhass.utils.get_logger(fun_name: str, emhass_conf: dict[str, Path], save_to_file: bool = True, logging_level: str = 'DEBUG') → tuple[Logger, StreamHandler]

Create a simple logger object.

Parameters:

• fun_name (str) – The Python function object name where the logger will be used

• emhass_conf (dict) – Dictionary containing the needed emhass paths

• save_to_file (bool, optional) – Write log to a file, defaults to True

Returns:

The logger object and the handler

Return type:

object

emhass.utils.get_root(file: str, num_parent: int = 3) → str

Get the root absolute path of the working directory.

Parameters:

• file – The passed file path with __file__

• num_parent (int, optional) – The number of parents levels up to desired root folder

Returns:

The root path

Return type:

str

emhass.utils.get_yaml_parse(params: str | dict, logger: Logger) → tuple[dict, dict, dict]

Perform parsing of the params into the configuration catagories

Parameters:

• params (str or dict) – Built configuration parameters

• logger (logging.Logger) – The logger object

Returns:

A tuple with the dictionaries containing the parsed data

Return type:

tuple(dict)

emhass.utils.handle_nan_values(df: DataFrame, handle_nan: str, timestamp_col: str, logger: Logger) → DataFrame

Handle NaN values in DataFrame according to specified strategy.

Parameters:

• df (pd.DataFrame) – Input DataFrame

• handle_nan (str) – Strategy for handling NaN values

• timestamp_col (str) – Name of timestamp column to exclude from processing

• logger (logging.Logger) – Logger object

Returns:

DataFrame with NaN values handled

Return type:

pd.DataFrame

emhass.utils.log_runtime_banner(logger, optim_conf: dict | None = None)

Log a single INFO line with EMHASS/Python/CVXPY/platform info for bug-report reproducibility.

When optim_conf is provided, the configured solver (or the EMHASS default Highs when the key is absent) is shown — this matches the solver the LP actually uses (see optimization.py constructor). When optim_conf is missing entirely (early-fail paths), falls back to cvxpy.installed_solvers()[0].

emhass.utils.normalize_heat_cool_mode(value: str, *, field_name: str = 'mode', context: str | None = None) → str

Normalize heat/cool mode values and raise on invalid input.

emhass.utils.param_to_config(param: dict[str, dict], logger: Logger) → dict[str, str]

A function that extracts the parameters from param back to the config.json format. Extracts parameters from config catagories. Attempts to exclude secrets hosed in retrieve_hass_conf.

Parameters:

• params – Built configuration parameters

• logger (logging.Logger) – The logger object

Returns:

The built config dictionary

Return type:

dict[str, str]

emhass.utils.parse_export_time_range(start_time: str, end_time: str | None, time_zone: <property object at 0x735b24c61530>, logger: Logger) → tuple[Timestamp, Timestamp] | tuple[bool, bool]

Parse and validate start_time and end_time for export operations.

Parameters:

• start_time (str) – Start time string in ISO format

• end_time (str | None) – End time string in ISO format (optional)

• time_zone (pd.Timestamp.tz) – Timezone for localization

• logger (logging.Logger) – Logger object

Returns:

Tuple of (start_dt, end_dt) or (False, False) on error

Return type:

tuple[pd.Timestamp, pd.Timestamp] | tuple[bool, bool]

emhass.utils.resample_and_filter_data(df: DataFrame, start_dt: Timestamp, end_dt: Timestamp, resample_freq: str, logger: Logger) → DataFrame | bool

Filter DataFrame to time range and resample to specified frequency.

Parameters:

• df (pd.DataFrame) – Input DataFrame with datetime index

• start_dt (pd.Timestamp) – Start datetime for filtering

• end_dt (pd.Timestamp) – End datetime for filtering

• resample_freq (str) – Resampling frequency string (e.g., ‘1h’, ‘30min’)

• logger (logging.Logger) – Logger object

Returns:

Resampled DataFrame or False on error

Return type:

pd.DataFrame | bool

emhass.utils.resolve_min_temperatures(config: dict, outdoor_temperature_forecast: ndarray | Series | list | None, length: int) → list[float]

Compute the effective per-slot lower temperature bound for a storage tank.

Combines two sources, taking the element-wise max so the more conservative floor always wins:

1. Static min_temperatures (or min_temperature) list - an absolute weather-independent floor for safety / comfort. Always respected.

2. min_temperature_curve dict - a weather-compensated floor matching the radiator emission law. Same shape as heating_curve: T = clip(offset - slope * T_outdoor, min_supply, max_supply).

When only one is set, that one is used. When neither is set, returns an empty list (caller decides how to react).

Parameters:

• config – Tank or thermal_battery dict potentially carrying min_temperatures and/or min_temperature_curve.

• outdoor_temperature_forecast – Per-slot outdoor temperature. May be None when only the static floor is configured.

• length – Optimization horizon length.

Returns:

List of per-slot minimum temperatures (length = length).

emhass.utils.resolve_thermal_battery_cop(hc: dict, outdoor_temperature_forecast: Sequence[float] | ndarray | Series | None, length: int | None = None) → ndarray

Resolve the per-timestep energy-conversion factor for a thermal_battery heat source.

The thermal_battery model treats the conversion factor uniformly as a COP-like multiplier on input power: Q_thermal = COP * P_in / 1000 * dt. Two source types are supported:

• Heat pump (default): COP is computed via the Carnot formula and varies with

  the outdoor temperature. Requires supply_temperature in hc; uses carnot_efficiency (default 0.4). If sense is set to “cool”, cooling Carnot lift is used (T_outdoor - T_supply).

• Constant-efficiency source (gas boiler, oil burner, district heating, etc.): efficiency in hc is used as a flat conversion factor for every timestep. supply_temperature is not required and outdoor temperature is ignored. The constant value passes through Carnot bounds intentionally (typical 0.85-0.95 for combustion sources).

Parameters:

• hc – The thermal_battery sub-config dict from def_load_config.

• outdoor_temperature_forecast – Outdoor temperature forecast in degrees Celsius. Required in heat-pump mode. In constant-efficiency mode it is ignored and may be None provided length is given explicitly.

• length – Number of timesteps in the returned array. Mandatory when outdoor_temperature_forecast is None; otherwise truncates or passes through the forecast length when set, or returns the full forecast length when unset.

Returns:

Numpy array of conversion factors, one per timestep.

emhass.utils.set_df_index_freq(df: DataFrame) → DataFrame

Set the freq of a DataFrame DateTimeIndex.

Parameters:

df (pd.DataFrame) – Input DataFrame

Returns:

Input DataFrame with freq defined

Return type:

pd.DataFrame

emhass.utils.stage_timer(stage_times: dict, name: str, logger: Logger | None = None)

Record wall-clock elapsed time of a block in stage_times[name].

Emits one DEBUG line per stage when logger is provided. Works across await because time.perf_counter() is a plain monotonic read.

async emhass.utils.treat_runtimeparams(runtimeparams: str, params: dict[str, dict], retrieve_hass_conf: dict[str, str], optim_conf: dict[str, str], plant_conf: dict[str, str], set_type: str, logger: Logger, emhass_conf: dict[str, Path]) → tuple[str, dict[str, dict]]

Treat the passed optimization runtime parameters.

Parameters:

• runtimeparams (str) – Json string containing the runtime parameters dict.

• params (str) – Built configuration parameters

• retrieve_hass_conf (dict) – Config dictionary for data retrieving parameters.

• optim_conf (dict) – Config dictionary for optimization parameters.

• plant_conf (dict) – Config dictionary for technical plant parameters.

• set_type (str) – The type of action to be performed.

• logger (logging.Logger) – The logger object.

• emhass_conf (dict) – Dictionary containing the needed emhass paths

Returns:

Returning the params and optimization parameter container.

Return type:

Tuple[str, dict]

emhass.utils.update_params_with_ha_config(params: str, ha_config: dict) → dict

Update the params with the Home Assistant configuration.

Parameters

paramsstr

The serialized params.

ha_configdict

The Home Assistant configuration.

Returns

dict

The updated params.