Pyscripts
Actual Brain¶
This part contains the scripts organized by the entities which they use. First are the EMHASS calls and the SOLAX export limit since they share lot of variables, split into 3 files. Second is the strategy selector. Please create a folder /config/pyscript with a subfolder /config/pyscript/modules and put the files there:
/config/pyscript/em.py - the triggers, the calls to EMHASS and the publish watchdog
/config/pyscript/modules/em_inputs.py - the entity names and the reads from / writes to Home Assistant
/config/pyscript/modules/em_functions.py - the logic, plain Python without Home Assistant or EMHASS
/config/pyscript/strategy.py - the strategy selector
Pyscript only imports from the modules folder, so the two em_* files have to be there.
Changing a module reloads em.py automatically.
There is a todo -> expose the adjustment of the spot_prices to the user by a entity.
EMHASS calls¶
There is not much here also. Just calls to the EMHASS day-ahead optimization and the EMHASS MPC. Please take a look at https://emhass.readthedocs.io/en/latest/advanced_math_model.html#the-emhass-optimizations nice grap.
Goal of these is to provide a plan that updates as the time goes by. You will need to probably adapt the names of the used entities to match your entities.
However, these are ready at the top of em_inputs.py, so it should be relatively straightforward.
The triggers. Each one reads the values with em_inputs, builds an Optimization result with em_functions
(the EMHASS action, the payload and the reason when it is skipped) and posts it to EMHASS.
The logging of the Optimization and of the EMHASS answer is commented out, uncomment it in em.py to find issues with the function code.
A failed call logs the ERROR and WARNING lines of the EMHASS answer.
The day-ahead runs at 23:00, to test it during the day call the pyscript.dayahead action in
Developer Tools -> Actions. EMHASS plans the day-ahead always for the next 24 hours and ignores prediction_horizon,
so it needs the prices of tomorrow. They are published in the afternoon, before that the day-ahead is skipped
and the log says why. It is not a dry run, the result is published and the strategy follows it until the next MPC at :30:
/config/pyscript/em.py
"""Triggers of the EMHASS optimizations, the publish watchdog and the Solax export limit.
The pyscript decorators and the calls to EMHASS live here. The values are read
in em_inputs.py, the logic is in em_functions.py, both in the modules folder.
"""
import aiohttp
import em_functions
import em_inputs
import logging
logger = logging.getLogger("spot_eval.pyscript")
# Not localhost: it resolves to IPv6 ::1 first, where port 5000 accepts and resets the connection
EMHASS_URL = "http://127.0.0.1:5000"
async def call_emhass(action, json):
"""Post to emhass endpoint with json."""
async with aiohttp.ClientSession() as session:
# Not async with: pyscript calls __aexit__ also when the post itself fails, aiohttp then
# raises an AttributeError ('_resp') hiding the real error. Closing the session releases it.
response = await session.post(
f"{EMHASS_URL}/action/{action}",
json=json, headers={"Content-Type": "application/json"}
)
# EMHASS answers with plain text, not json
text = await response.text()
if response.status >= 400:
# Not a ClientResponseError, its repr holds the whole request and the whole EMHASS log
raise RuntimeError(f"EMHASS answered {response.status}: {em_functions.emhass_errors(text)}")
return text
async def publish():
"""Publish the last optimization to Home Assistant, store the time of an ok publish.
Ok means EMHASS answered without an error status. Explicit repr(), pyscript
ignores the !r conversion in f-strings.
"""
try:
result = await call_emhass(em_functions.PUBLISH, {})
except Exception as e:
logger.error(f"{em_functions.PUBLISH} failed {repr(e)}")
return
# debug, this runs every 5 minutes
# logger.debug(f"{em_functions.PUBLISH} result {repr(result)}")
em_inputs.set_last_publish(em_inputs.now())
async def run_optimization(optimization):
"""Post the optimization to EMHASS and publish the result."""
if optimization.skipped:
logger.info(f"Skipping {optimization.action}: {optimization.skipped}")
return
try:
result = await call_emhass(optimization.action, optimization.payload)
except Exception as e:
logger.error(f"{optimization.action} failed {repr(e)}")
return
# logger.info(f"{optimization.action} result {repr(result)}")
await publish()
logger.info(f"Finished the {optimization.action}")
# Also a service, pyscript.dayahead, to run it during the day for testing
@service
@time_trigger("cron(00 23 * * *)")
async def dayahead():
"""Perform Dayahead optimization."""
optimization = em_functions.dayahead_optimization(
em_inputs.now(),
em_inputs.buy_price_attributes(),
em_inputs.sell_price_attributes(),
em_inputs.pv_watts, # the reader itself, called only when the optimization runs
em_inputs.battery_soc(),
em_inputs.tomorrow_production(),
)
# The result of the function code, to find issues with it later
# logger.info(f"Optimization {repr(optimization)}")
await run_optimization(optimization)
@time_trigger("cron(30 * * * *)")
async def mpc():
"""This is the optimization happening mid day"""
optimization = em_functions.mpc_optimization(
em_inputs.now(),
em_inputs.buy_price_attributes(),
em_inputs.sell_price_attributes(),
em_inputs.pv_watts, # the reader itself, called only when the optimization runs
em_inputs.battery_soc(),
em_inputs.tomorrow_production(),
)
# The result of the function code, to find issues with it later
# logger.info(f"Optimization {repr(optimization)}")
await run_optimization(optimization)
@time_trigger("startup", "cron(*/5 * * * *)")
async def publish_every_5_minutes():
"""Keep the EMHASS sensors in Home Assistant current, the watchdog checks it."""
await publish()
# 2 minutes after the publish, so the check sees the publish of the same 5 minutes.
# Not at startup: until the first check after a restart the binary sensor doesn't exist and
# the Solax automations wait, by then the startup publish is done.
@time_trigger("cron(2-59/5 * * * *)")
async def publish_watchdog():
"""Check that the last ok publish is less than 15 minutes old."""
last_publish = em_inputs.last_publish()
status, age_minutes = em_functions.publish_status(em_inputs.now(), last_publish)
previous = em_inputs.publish_watchdog_status()
em_inputs.set_publish_watchdog(status, age_minutes, last_publish)
# Log only the change, not every 5 minutes
if status == previous:
return
if status == em_functions.PUBLISH_STALE:
if age_minutes is None:
logger.warning("EMHASS publish is stale, there was no ok publish yet")
else:
logger.warning(f"EMHASS publish is stale, last ok publish {last_publish}, {age_minutes} minutes ago")
elif previous == em_functions.PUBLISH_STALE:
logger.info(f"EMHASS publish is ok again, last ok publish {last_publish}")
@state_trigger("True", watch=[em_inputs.SENSOR_SELL_PRICE])
def enable_disable_solax_grid_export():
"""Enable or disable the export based on your sell price"""
previous_limit = em_inputs.export_limit()
spot_price = em_inputs.spot_sell_price()
logger.debug(f"Previous value {previous_limit} - {float(spot_price)} ")
out, allowed = em_functions.export_limit(
spot_price,
em_inputs.sell_threshold(),
em_inputs.restore_export_limit(),
)
em_inputs.set_export_allowed(allowed)
# to avoid storing to EEPROM on each evaluation
if previous_limit != out:
logger.debug(f"Changing solax export limit from {previous_limit} to {out}")
em_inputs.set_export_limit(out)
else:
logger.debug(f"No change to solax export limit {previous_limit} - {out}")
The entity names and everything touching Home Assistant. This is the file to adapt to your entities:
/config/pyscript/modules/em_inputs.py
"""Reads from and writes to Home Assistant.
Everything that touches Home Assistant lives here: the entity names, the state
reads and the service calls writing to Solax.
The logic that works with the values is in em_functions.py, the triggers and
the calls to EMHASS in em.py.
Adapt the entity names below to match your entities.
"""
from datetime import datetime
from em_functions import PUBLISH_OK, PUBLISH_STALE, TIMEZONE
# This is the OTE integation sensor, if you use something else, please adapt the spot price functions
SENSOR_TODAY_SPOT = "sensor.current_spot_electricity_price_15min"
SENSOR_TODAY_SPOT_BUY = "sensor.current_spot_electricity_price_15min"
SENSOR_TODAY_SPOT_SELL = "sensor.current_spot_electricity_price_15min"
# Weather integration
SENSOR_TODAY_WEATHER = "sensor.energy_production_today_2"
SENSOR_TOMORROW_WEATHER = "sensor.energy_production_tomorrow_2"
# Solax
SENSOR_BATTERY_CAPACITY = "sensor.solax_battery_capacity"
# Sensor containing the current spot *sell* price
# Used for the disable enable
SENSOR_SELL_PRICE = "sensor.current_spot_electricity_price"
# Solax field for changing the user limit
SENSOR_EXPORT_LIMIT = "number.solax_export_control_user_limit"
# Input sensor for saving the status
INPUT_SENSOR_EXPORT_ALLOWED = "input_boolean.export_allowed"
# Input sensor for modifying the minimum sell price
INPUT_SENSOR_SELL_THRESHOLD = "input_number.export_price_threshold"
# Input sensor that defines the maximum limit restoration
INPUT_SENSOR_RESTORE_EXPORT_ALLOWED = "input_number.maximum_export_allowed"
# Entities written by em.py. Neither is kept across Home Assistant restarts, on purpose:
# the EMHASS sensors are gone after a restart until the first ok publish, so a publish
# from before the restart must not count.
# Time of the last ok publish to EMHASS, as an ISO timestamp
PYSCRIPT_LAST_PUBLISH = "pyscript.emhass_last_publish"
# Result of the publish watchdog as a problem binary sensor: on when the publish is stale,
# off when it is ok, with the age as an attribute. The Solax automations act only when off.
BINARY_SENSOR_PUBLISH_STALE = "binary_sensor.emhass_publish_stale"
def now():
"""Current time in the spot market timezone."""
return datetime.now(TIMEZONE)
def buy_price_attributes():
"""{iso time: price} forecast of the buy spot price."""
return hass.states.get(SENSOR_TODAY_SPOT_BUY).attributes
def sell_price_attributes():
"""{iso time: price} forecast of the sell spot price."""
return hass.states.get(SENSOR_TODAY_SPOT_SELL).attributes
def pv_watts():
"""{iso time: watts} production forecast of today and tomorrow."""
today_attributes = hass.states.get(SENSOR_TODAY_WEATHER).attributes
tomorrow_attributes = hass.states.get(SENSOR_TOMORROW_WEATHER).attributes
return today_attributes["watts"] | tomorrow_attributes["watts"]
def battery_soc():
"""Battery level in percent, as the raw state."""
return hass.states.get(SENSOR_BATTERY_CAPACITY).state
def tomorrow_production():
"""Forecasted production tomorrow in kWh, as the raw state."""
return hass.states.get(SENSOR_TOMORROW_WEATHER).state
def spot_sell_price():
"""Current spot sell price, as the raw state."""
return hass.states.get(SENSOR_SELL_PRICE).state
def sell_threshold():
"""User defined minimum sell price, as the raw state."""
return hass.states.get(INPUT_SENSOR_SELL_THRESHOLD).state
def restore_export_limit():
"""User defined export limit to restore when exporting is allowed, as the raw state."""
return hass.states.get(INPUT_SENSOR_RESTORE_EXPORT_ALLOWED).state
def export_limit():
"""Export limit currently set in Solax, as a number."""
return float(hass.states.get(SENSOR_EXPORT_LIMIT).state)
def set_export_allowed(allowed):
"""Store whether exporting is allowed in the input boolean."""
if allowed:
service.call("input_boolean", "turn_on", entity_id=INPUT_SENSOR_EXPORT_ALLOWED)
else:
service.call("input_boolean", "turn_off", entity_id=INPUT_SENSOR_EXPORT_ALLOWED)
def set_export_limit(value):
"""Write the export limit to Solax."""
service.call("number", "set_value", entity_id=SENSOR_EXPORT_LIMIT, value=value)
def last_publish():
"""Time of the last ok publish, as the raw state (an ISO timestamp or unknown)."""
# Doesn't exist until the first ok publish after a restart
if not state.exist(PYSCRIPT_LAST_PUBLISH):
return "unknown"
return state.get(PYSCRIPT_LAST_PUBLISH)
def set_last_publish(time):
"""Store the time of an ok publish."""
state.set(PYSCRIPT_LAST_PUBLISH, time.isoformat())
def publish_watchdog_status():
"""Last result of the publish watchdog: PUBLISH_OK, PUBLISH_STALE, or None before its first run."""
# Doesn't exist until the first run after a restart
if not state.exist(BINARY_SENSOR_PUBLISH_STALE):
return None
value = state.get(BINARY_SENSOR_PUBLISH_STALE)
if value == "on":
return PUBLISH_STALE
if value == "off":
return PUBLISH_OK
return None
def set_publish_watchdog(status, age_minutes, last_publish):
"""Store the result of the publish watchdog, usable in automations and dashboards.
device_class problem makes Home Assistant show on as Problem and off as OK.
"""
state.set(
BINARY_SENSOR_PUBLISH_STALE,
"on" if status == PUBLISH_STALE else "off",
device_class="problem",
friendly_name="EMHASS publish",
age_minutes=age_minutes,
last_publish=str(last_publish),
)
The logic. Plain Python, it is covered by the unit tests in the tests folder of the repository
(python3 -m unittest discover -s tests):
/config/pyscript/modules/em_functions.py
"""Pure logic for the EMHASS optimizations and the Solax export limit.
Only the standard library is used here, nothing from Home Assistant, pyscript or
EMHASS, so this file can be imported and unit tested outside of Home Assistant.
The optimizations are returned as an Optimization result, em.py posts it to EMHASS.
The Home Assistant reads and writes live in em_inputs.py, the triggers in em.py.
"""
import json
import logging
import re
from collections import namedtuple
from zoneinfo import ZoneInfo
from datetime import datetime, timedelta, timezone
logger = logging.getLogger("spot_eval.pyscript")
# EMHASS actions, the endpoint of http://127.0.0.1:5000/action/<action>
MPC = "naive-mpc-optim"
DAYAHEAD = "dayahead-optim"
PUBLISH = "publish-data"
# Result of building an optimization, em.py posts the payload to the EMHASS action.
# A namedtuple so the repr works inside pyscript, where special methods of own classes don't.
# action: the EMHASS action to call
# payload: the json to post, None when skipped
# skipped: why the optimization is skipped, None when it should run
Optimization = namedtuple("Optimization", ["action", "payload", "skipped"], defaults=[None])
# Timezone of the spot market, the dayahead optimization runs at 23:00 local time
TIMEZONE = ZoneInfo("Europe/Prague")
# Adjustments added to the raw spot price (distribution fees, taxes, trader margin)
BUY_PRICE_ADJUSTMENT = 0.5 + 2.5
SELL_PRICE_ADJUSTMENT = -0.5
# maxim amount of the lines 2
LINES_LIMIT = 24 * 4
# A forecast value older than this means the forecast data is missing
MAX_FORECAST_AGE = timedelta(hours = 2)
# Hour of the day the dayahead optimization runs, the MPC optimizes until the next one
DAYAHEAD_HOUR = 23
# Below this amount of 15 minute lines the MPC is skipped as per the emhass docs
MIN_MPC_LINES = 6
# Tomorrow's production (kWh) below this means the battery should end the day fuller
LOW_PRODUCTION_THRESHOLD = 20
SOC_FINAL_LOW_PRODUCTION = 0.7
SOC_FINAL_NORMAL_PRODUCTION = 0.5
def spot_forecast(attributes):
"""(time, price) of the spot price sensor attributes named by an ISO time.
The other attributes are skipped: unit_of_measurement, icon, friendly_name and
whatever else the sensor has, e.g. state_class or device_class.
"""
forecast = []
for k, v in attributes.items():
try:
date = datetime.fromisoformat(k)
except ValueError:
continue
forecast += [(date, v)]
# Otherwise a sensor with differently named attributes just gives no prices
if not forecast:
logger.warning(f"No spot prices in the attributes {list(attributes)}")
return forecast
def get_spot_prices_day_ahead(now, attributes, price_adjustment):
"""Get the prices from the forecast."""
prices = []
for date, v in spot_forecast(attributes):
v_a = v
if date < now:
continue
prices += [v_a + price_adjustment]
if len(prices) == LINES_LIMIT:
break
return prices
def get_spot_prices_mpc(attributes, lines_limit, start, tomorrow, price_adjustment):
"""Get the spot prices for the mpc."""
prices = []
for date, v in spot_forecast(attributes):
v_a = v
if start <= date < tomorrow:
prices += [v_a + price_adjustment ]
if len(prices) == lines_limit:
break
return prices
def to_quarter_hour(time):
"""Get the start of the quarter hour containing time, in UTC."""
time = time.astimezone(timezone.utc)
return time.replace(minute=time.minute - time.minute % 15, second=0, microsecond=0)
def to_nice_minutes(values, now):
"""Get LINES_LIMIT 15 minute values from a {time: value} forecast, starting at the quarter hour of now.
Each value is held until the next time in the forecast. When the last value is older
than MAX_FORECAST_AGE, e.g. the next day is missing, the list ends early.
"""
forecast = sorted([(datetime.fromisoformat(k), v) for k, v in values.items()])
# Step in UTC, adding minutes to the local time breaks on the DST change
current = to_quarter_hour(now)
pv_forecast = []
index = 0
# No production before the forecast starts
value = 0
value_time = None
for _ in range(LINES_LIMIT):
# Take the last forecast value at or before current
while index < len(forecast) and forecast[index][0] <= current:
value_time, value = forecast[index]
index += 1
# Don't keep the old value when the forecast is missing, let the optimization fail
if value_time is not None and current - value_time > MAX_FORECAST_AGE:
logger.warning(f"PV forecast is missing after {value_time}, returning only {len(pv_forecast)} values")
break
pv_forecast += [value]
current = current + timedelta(minutes = 15)
return pv_forecast
def soc_final(tomorrow_production):
"""Battery level to end the optimization with, fuller when tomorrow's production is low."""
if float(tomorrow_production) < LOW_PRODUCTION_THRESHOLD:
return SOC_FINAL_LOW_PRODUCTION
return SOC_FINAL_NORMAL_PRODUCTION
def mpc_horizon_end(now):
"""The next dayahead optimization at DAYAHEAD_HOUR local time, the MPC optimizes until then."""
# After the dayahead at 23:00 optimize until the next one
if now.hour >= DAYAHEAD_HOUR:
t = now + timedelta(days=1)
else:
t = now
return datetime(year=t.year, month=t.month, day=t.day, hour=DAYAHEAD_HOUR, tzinfo=TIMEZONE)
def mpc_lines_limit(start, end):
"""Number of 15 minute steps between start and end.
Start is in UTC so the DST change is counted in.
"""
return int((end - start).total_seconds() // (15 * 60))
def dayahead_optimization(now, buy_attributes, sell_attributes, pv_watts, soc_init, tomorrow_production):
"""Build the dayahead optimization, skipped when the prices don't cover the whole day.
buy_attributes, sell_attributes: {iso time: price} attributes of the spot price sensors
pv_watts: function returning the {iso time: watts} forecast of today and tomorrow,
only called when the optimization runs
soc_init: battery level in percent
tomorrow_production: forecasted production tomorrow in kWh
EMHASS plans the dayahead always for the next LINES_LIMIT steps, it ignores
prediction_horizon and rejects a shorter forecast. Run during the day, e.g. for
testing, it works only after the prices of tomorrow are published in the afternoon.
The pv forecast is read after the skip decision, like in mpc_optimization.
"""
# Same first slot as the pv forecast
start = to_quarter_hour(now)
load_cost_forecast = get_spot_prices_day_ahead(start, buy_attributes, BUY_PRICE_ADJUSTMENT)
prod_price_forecast = get_spot_prices_day_ahead(start, sell_attributes, SELL_PRICE_ADJUSTMENT)
lines = min(len(load_cost_forecast), len(prod_price_forecast))
if lines < LINES_LIMIT:
return Optimization(DAYAHEAD, None, f"only {lines} lines of prices from {start.isoformat()}, "
f"the dayahead needs {LINES_LIMIT}, the prices of tomorrow are published in the afternoon")
return Optimization(DAYAHEAD, {
"pv_power_forecast" : to_nice_minutes(pv_watts(), now),
"load_cost_forecast" : load_cost_forecast,
"prod_price_forecast" : prod_price_forecast,
"soc_init": float(soc_init) / 100,
"soc_final": soc_final(tomorrow_production),
})
def mpc_optimization(now, buy_attributes, sell_attributes, pv_watts, soc_init, tomorrow_production):
"""Build the MPC optimization, skipped when the horizon is too short.
Takes the same inputs as dayahead_optimization. Every forecast is cut to the same
prediction_horizon, EMHASS needs them exactly that long. The pv forecast is read
after the skip decision, the forecast sensors may be unavailable late in the day.
"""
start = to_quarter_hour(now)
end = mpc_horizon_end(now)
lines_limit = mpc_lines_limit(start, end)
load_cost_forecast = get_spot_prices_mpc(buy_attributes, lines_limit, start, end, BUY_PRICE_ADJUSTMENT)
prod_price_forecast = get_spot_prices_mpc(sell_attributes, lines_limit, start, end, SELL_PRICE_ADJUSTMENT)
# EMHASS needs every forecast exactly prediction_horizon long, the prices may end sooner
lines_limit = min(lines_limit, len(load_cost_forecast), len(prod_price_forecast))
# Early return we do not perform optimizations as per the emhass docs if the ending of the day is close.
if lines_limit < MIN_MPC_LINES:
return Optimization(MPC, None, f"only {lines_limit} lines until {end.isoformat()}, less than {MIN_MPC_LINES}")
return Optimization(MPC, {
"pv_power_forecast" : to_nice_minutes(pv_watts(), now)[:lines_limit],
"load_cost_forecast" : load_cost_forecast[:lines_limit],
"prod_price_forecast" : prod_price_forecast[:lines_limit],
"soc_init": float(soc_init) / 100,
"soc_final": soc_final(tomorrow_production),
"prediction_horizon": lines_limit
})
def export_limit(spot_price, sell_threshold, restore_limit):
"""Decide the Solax export limit from the sell price.
Returns (limit, allowed): the limit to set as a number and whether exporting is allowed.
The export is allowed when the spot price is above the user threshold.
"""
if float(spot_price) > float(sell_threshold):
return float(restore_limit), True
return 0, False
# Longest non-json EMHASS answer kept in an error, an internal server error is a whole html page
MAX_ERROR_LENGTH = 300
def emhass_errors(text):
"""The ERROR and WARNING lines of a failed EMHASS answer, joined with " | ".
EMHASS answers a failed action with its log lines as a json list. Without an ERROR or
WARNING line the last line is returned. Any other answer, e.g. the html page of an
internal server error, is returned without the html tags, cut to MAX_ERROR_LENGTH.
"""
try:
lines = json.loads(text)
except ValueError:
lines = None
if not isinstance(lines, list) or not lines:
return " ".join(re.sub("<[^>]+>", " ", text).split())[:MAX_ERROR_LENGTH]
errors = [str(line) for line in lines if str(line).startswith(("ERROR", "WARNING"))]
if not errors:
errors = [str(lines[-1])]
return " | ".join(errors)
# An ok publish older than this means EMHASS stopped publishing, em.py publishes every 5 minutes
MAX_PUBLISH_AGE = timedelta(minutes = 15)
PUBLISH_OK = "ok"
PUBLISH_STALE = "stale"
def publish_status(now, last_publish):
"""Decide whether the last ok publish is recent enough.
last_publish: the stored ISO timestamp of the last ok publish, as the raw state
Returns (status, age_minutes): PUBLISH_OK when the publish is less than MAX_PUBLISH_AGE
old, PUBLISH_STALE otherwise. age_minutes is None when there is no usable timestamp,
e.g. never published yet, which counts as stale.
"""
try:
published = datetime.fromisoformat(str(last_publish))
except ValueError:
return PUBLISH_STALE, None
# A timestamp without timezone can't be compared to now
if published.tzinfo is None:
return PUBLISH_STALE, None
age = now - published
age_minutes = round(age.total_seconds() / 60, 1)
if age < MAX_PUBLISH_AGE:
return PUBLISH_OK, age_minutes
return PUBLISH_STALE, age_minutes
Publish watchdog¶
em.py publishes the EMHASS results to Home Assistant every 5 minutes and right after each optimization.
Every publish EMHASS answers without an error stores its time in pyscript.emhass_last_publish.
2 minutes after each publish a check compares that time to now. Less than 15 minutes old is ok, otherwise stale.
The result is the binary_sensor.emhass_publish_stale problem sensor: off (shown as OK) when ok, on (shown as Problem) when stale,
with the age in minutes as the age_minutes attribute. The Solax automations act only while it is off.
It is created by pyscript without a unique ID, so you can't rename it or assign it an area in the UI.
The log gets a warning when it turns stale and an info when it is ok again.
Neither entity is kept across a Home Assistant restart, on purpose. The EMHASS sensors are gone after a restart until the first publish, so a publish from before the restart must not count. Until the first check after the restart the binary sensor doesn't exist and the automations wait.
EMHASS answers ok even when it could not write the sensors to Home Assistant, check its log in that case.
/config/pyscript/em.py
# 2 minutes after the publish, so the check sees the publish of the same 5 minutes.
# Not at startup: until the first check after a restart the binary sensor doesn't exist and
# the Solax automations wait, by then the startup publish is done.
@time_trigger("cron(2-59/5 * * * *)")
async def publish_watchdog():
"""Check that the last ok publish is less than 15 minutes old."""
last_publish = em_inputs.last_publish()
status, age_minutes = em_functions.publish_status(em_inputs.now(), last_publish)
previous = em_inputs.publish_watchdog_status()
em_inputs.set_publish_watchdog(status, age_minutes, last_publish)
# Log only the change, not every 5 minutes
if status == previous:
return
if status == em_functions.PUBLISH_STALE:
if age_minutes is None:
logger.warning("EMHASS publish is stale, there was no ok publish yet")
else:
logger.warning(f"EMHASS publish is stale, last ok publish {last_publish}, {age_minutes} minutes ago")
elif previous == em_functions.PUBLISH_STALE:
logger.info(f"EMHASS publish is ok again, last ok publish {last_publish}")
Solax enable disable¶
This section is already included in the previous files, but i would just highlight it here. We are taking look at the spot price and comparing it to user defined sell threshold. Based on that we set the export limit in solax. Each write to the solax is written to write-limited (100 000 - you probably know), but this happens 2x or 4 times a day max.
The decision:
/config/pyscript/modules/em_functions.py
def export_limit(spot_price, sell_threshold, restore_limit):
"""Decide the Solax export limit from the sell price.
Returns (limit, allowed): the limit to set as a number and whether exporting is allowed.
The export is allowed when the spot price is above the user threshold.
"""
if float(spot_price) > float(sell_threshold):
return float(restore_limit), True
return 0, False
The trigger writing it to solax:
/config/pyscript/em.py
@state_trigger("True", watch=[em_inputs.SENSOR_SELL_PRICE])
def enable_disable_solax_grid_export():
"""Enable or disable the export based on your sell price"""
previous_limit = em_inputs.export_limit()
spot_price = em_inputs.spot_sell_price()
logger.debug(f"Previous value {previous_limit} - {float(spot_price)} ")
out, allowed = em_functions.export_limit(
spot_price,
em_inputs.sell_threshold(),
em_inputs.restore_export_limit(),
)
em_inputs.set_export_allowed(allowed)
# to avoid storing to EEPROM on each evaluation
if previous_limit != out:
logger.debug(f"Changing solax export limit from {previous_limit} to {out}")
em_inputs.set_export_limit(out)
else:
logger.debug(f"No change to solax export limit {previous_limit} - {out}")
Strategy selector¶
Honestly this script could be improved quite a bit, but it works for me. I will probably do once I have some energy.
Before trying to understand this, please take a look at the outputs from EMHASS. Here we are taking a look at the variables the EMHASS give us: https://emhass.readthedocs.io/en/latest/advanced_math_model.html#the-emhass-optimizations This script updates the strategy entity.
This is the script number 2:
/config/pyscript/strategy.py
import logging
logger = logging.getLogger("spot_eval.pyscript")
## Do not change unless you know what you are doing
SELL_TO_GRID = "Sell to grid"
CHARGE = "Charge"
DISCHARGE = "Discharge"
NO_ACTION = "No action"
# Change to the forecast sensors from emhass.
SENSOR_PV_FORECAST = "sensor.p_pv_forecast"
SENSOR_P_LOAD_FORECAST = "sensor.p_load_forecast"
SENSOR_P_BATT_FORECAST = "sensor.p_batt_forecast"
SENSOR_P_GRID_FORECAST = "sensor.p_grid_forecast"
# Change to the solax battery
SENSOR_BATTERY_CAPACITY = "sensor.solax_battery_capacity"
# INPUT_SELECTOR used to save the strategy
INPUT_INVERTOR_STRATEGY = "input_select.invertor_strategy"
@state_trigger("1 == 1",
watch=[SENSOR_PV_FORECAST, SENSOR_P_LOAD_FORECAST, SENSOR_P_BATT_FORECAST, SENSOR_P_GRID_FORECAST])
def select_strategy():
p_pv = float(hass.states.get(SENSOR_PV_FORECAST).state)
p_load = float(hass.states.get(SENSOR_P_LOAD_FORECAST).state)
p_grid = float(hass.states.get(SENSOR_P_GRID_FORECAST).state)
p_batt = float(hass.states.get(SENSOR_P_BATT_FORECAST).state)
soc = float(hass.states.get(SENSOR_BATTERY_CAPACITY).state)
strategy = select(p_pv, p_load, p_grid, p_batt, soc)
service.call("input_select", "select_option", entity_id=INPUT_INVERTOR_STRATEGY, option=strategy)
logger.debug("Strategy chosen")
# b_batt -> negative is charging, positive is discharge
# feel free to modify the strategy selection
def select(p_pv: int, p_load: int, p_grid: int, p_batt: int, soc: float) -> str:
# check for charging
if (p_batt < 0 and p_grid > 0) and (- p_batt > (0.5 * p_grid)):
return CHARGE
if (p_batt > 0 and p_grid < 0) and (p_batt > (0.5 * - p_grid)):
return DISCHARGE
if p_batt < 500 and p_grid < 0 and soc < 85:
return SELL_TO_GRID
return NO_ACTION