Home Assistant sensor for Marstek Venus battery storage

Marstek_Venus_Signal.yaml

- sensor:
  - name: "Marstek Quellsignal"
    # Generate a hate-driven differential signal for the Marstek Venus E battery storage.
    # We hate both throwing away energy into the grid and buying energy from the grid, but we hate one more than the other.
    # By quantifying our bad feelings we can prioritize what to do if it's hard to reach zero.
    # The trade-off becomes mostly relevant if the power oscillates a lot and we try to tame this oscillation by calmly accepting
    # that less countersteering is better. Or if we move the target power to a non-zero value, because we anticipate future power jumps.
    #
    # The storage interprets the value as the difference between it's current power output (which I don't know) and the power required by the home.
    # So this is the meaning of several example values from the perspective of the storage:
    #
    # [value]     [meaning for the storage]
    #       0     I should hold my current power value exactly the way it is now
    #    -100     We are drawing 100W from the grid. I should increase (decrease) my output (input) by 100W to balance this out.
    #     +42     We are throwing 42W into the grid. I should decrease (increase) my output (input) by 42W to balance this out.
    unit_of_measurement: "W"
    device_class: power
    state_class: measurement
    icon: "mdi:controller"
    unique_id: marstek_quellsignal
    state: >
      {# -------------------------------------
          All sensors for this template
         ------------------------------------- #}
      {% set power_into_grid_actual   = -(states("sensor.tibber_local_0100100700ff")            | float(0))%}   {# [-], because tibber reports power from grid, but we must provide the power into the grid #}
      {% set last_updated             =   states.sensor.tibber_local_0100100700ff.last_updated %}               {# this template must fire periodically in order to get outdated timestamps, too #}
      {% set grid_std_dev             =   states("sensor.standardabweichung_netzleistung_1min") | float(0) %}   {# standard deviation of the grid power [W] #}
      {% set energy_to_grid_recently  =   states("sensor.eingespeiste_energie_in_1min")         | float(0) %}   {# [Wh] #}
      {% set energy_from_grid_recently=   states("sensor.entnommene_energie_in_1min")           | float(0) %}   {# [Wh] #}
      {% set solar_power              =   states("sensor.solar_leistung")                       | float(0) %}   {# total solar power [W] #}
      {% set solar_power_stddev       =   states("sensor.standardabweichung_solarleistung")     | float(0) %}   {# standard deviation of the solar power [W] #}
      {% set enable_unimeter_quirks   =   not (states("binary_sensor.speicher_entladen_sinnvoll") | bool(true)) %}  {# this is usually false if uni-meter shall be used in regular mode! #}
      {% set price_level              =   state_attr("sensor.meine_heimatstrasse_42_strompreis", 'price_level') | upper %}   {# Tibber price sensor (LOW, NORMAL, HIGH) #}
      {% set _ignore_me               =   states("sensor.dummy_signal_every_second") %}                         {# forcefully update this template at least(!) once a second #}
      
      {# -------------------------------------
          All configuration variables
         ------------------------------------- #}
      {# Quantify our bad feelings in regards to feeding into the grid vs. buying from the grid. Total hate should be 1 (= 100%) #}
      {% set HATE_FEED_IN               = 0.7 %}                {# how much we hate throwing energy away #}
      {% set HATE_RECEIVE               = 1 - HATE_FEED_IN %}   {# how much we hate buying energy #}
      {% set AGE_TOLERANCE_SEC          = 10 %}                 {# older values lack correctness [s] #}
      {% set UNIMETER_TOL_POWER_SHORT   = 150 %}                {# short-term tolerance for uni-meter's feature to prevent discharging the battery [W] #}
      {% set UNIMETER_TOL_POWER_LONG    = 50 %}                 {# long-term tolerance for uni-meter's feature to prevent discharging the battery [W] #}
      
      {# helper variables #}
      {% set HOURS_TO_SECONDS   = 3600 %}
      {% set age_of_measurement = as_timestamp(now())  -  as_timestamp(last_updated) %}
      {% set is_wasting_energy  = power_into_grid_actual > 0 %}
      {% set active_hate_factor = HATE_FEED_IN if is_wasting_energy else HATE_RECEIVE %}
      {% set prefer_receive     = HATE_FEED_IN > HATE_RECEIVE %}
      {% set offset_flat        = (    15   if price_level == "LOW"
                                  else 10   if price_level == "NORMAL"
                                  else  5   if price_level == "HIGH"
                                  else 10 ) * (2*prefer_receive - 1)  %} {# constant power offset [W] #}
      
      {#
        OFFSET 1:
        Dynamic offset based on the recently drawn or wasted energy
        Oh, the entire control loop has failed in the past? --> Try to offset us away from there
      #}
      {% set PERIOD_PAST_SEC      = 60 %}
      {% set mean_power_to_grid   = energy_to_grid_recently   * (HOURS_TO_SECONDS / PERIOD_PAST_SEC) %} {# always >= 0 [W] #}
      {% set mean_power_from_grid = energy_from_grid_recently * (HOURS_TO_SECONDS / PERIOD_PAST_SEC) %} {# always >= 0 [W] #}
      {% set offset_history_based = (   HATE_FEED_IN * mean_power_to_grid
                                      - HATE_RECEIVE * mean_power_from_grid) %}
      
      {# no countersteering against offset_flat #}
      {% if offset_history_based * offset_flat < 0 %}
        {% if offset_flat >= 0 %}
          {% set offset_history_based = min(offset_history_based + offset_flat, 0) %}   {# offset_flat > 0, offset_history_based < 0 #}
        {% else %}                
          {% set offset_history_based = max(offset_history_based + offset_flat, 0) %}   {# offset_flat < 0, offset_history_based > 0 #}
        {% endif %}
      {% endif %}
      {#
        We still prefer to *not increase* any deviations from zero even more:
        If we are on the right side of the power but below that offset, let's not change anything and wait for the next data value to arrive.
      #}
      {% if offset_history_based * power_into_grid_actual < 0 %}
        {% set offset_history_based = 0 %}
      {% endif %}
      
      
      {# 
        OFFSET 2:
        Dynamic offset based on the variability of solar power
      #}
      {% set STD_SOLAR_FAC =       2    if price_level == "LOW"
                              else 1.5  if price_level == "NORMAL"
                              else 1    if price_level == "HIGH"
                              else 1.5                            %} {# afaik, there is only LOW, NORMAL, HIGH #}
      {% set offset_solar_based = STD_SOLAR_FAC * (HATE_FEED_IN - HATE_RECEIVE) * solar_power_stddev %}
      
      
      {#
        OSCILLATION PREVENTION:
        Dynamic tolerance based on the recent power variability and price level
      
        Small constant tolerance (like 1W) if we are in the most-hated domain -> Get us away from there, quickly!
        Higher variability-scaled tolerance in the less-hated power hemisphere.
        The higher tolerance reduces oscillations efficiently, but reaching the zero line takes longer.
      #}
      {% set STD_GRID_FAC =       5 if price_level == "LOW"
                            else  4 if price_level == "NORMAL"
                            else  3 if price_level == "HIGH"
                            else  4                            %} {# afaik, there is only LOW, NORMAL, HIGH #}
      {% set power_tolerance =      1 if is_wasting_energy == prefer_receive
                              else 30 + STD_GRID_FAC * max(grid_std_dev - solar_power_stddev, 0) %}
      
      {% set a = 0.2 %}
      {% set linear_fac_power = a + (1-a) * min((power_into_grid_actual|abs)/power_tolerance, 1) %}     {# https://www.desmos.com/calculator/btbxe6cgdh #}
      {# set smoothing_gaussian_power = 1 - e**(-5 * ((power_into_grid_actual/power_tolerance)**2)) #}  {# https://www.desmos.com/calculator/5nm49lx8wz #}
      {% set smoothing_gaussian_age   =     e**(-5 * ((age_of_measurement/AGE_TOLERANCE_SEC)**4)) %}    {# https://www.desmos.com/calculator/kuhd5ashid #}
      
      {# Standard output calculation #}
      {% set output = (   offset_flat
                        + offset_history_based
                        + offset_solar_based
                        + power_into_grid_actual) * linear_fac_power * smoothing_gaussian_age %}
      
      {# ------------------------------------------------------------------------------------------------------------------------------------
        UNI_METER QUIRKS:
        If we have told uni-meter to prevent discharging the battery, we must be careful if we really allow the output to become negative,
        as negative means: we need power from the grid and want to discharge the battery now. uni-meter will then pause entirely to let the
        storage run idle. This should only happen for real and relevant consumers, not for our hacky offset or minor variations.
        ----------------------------------------------------------------------------------------------------------------------------------- #}
      {% if enable_unimeter_quirks %}
        {# quirk 1: ignore variable offsets (subtract what was added before) #}
        {% set output =    output
                        - (offset_flat + offset_history_based + offset_solar_based) * linear_fac_power * smoothing_gaussian_age %}
      
        {# quirk 2: "Dead zone" for small negative output: lie to the storage that everything is perfect as-is #}
        {% if     power_into_grid_actual <= 0
              and (-power_into_grid_actual) < UNIMETER_TOL_POWER_SHORT
              and mean_power_from_grid < UNIMETER_TOL_POWER_LONG %}
      
          {% set output = 0 %}  {# hey Venus, keep doing what you're doing! #}
        {% endif %}
      {% endif %}
      
      {{ output }}


- triggers:
  - trigger: time_pattern
    seconds: /1     # update once every second
  sensor:
    - name: "dummy signal every second"
      icon: "mdi:settings-helper"
      unique_id: dummy_signal_every_second
      state: >
        {{ 0 }}

Background info:

1. I have a medium balcony PV system and I get no money for feeding into the grid
   → It's impossible to be 100% self-sufficient, so it will always be necessary to receive grid power. And it's super dumb to feed into the grid, especially if it's from the battery.
   ⇒ If a choice is necessary, prefer to receive energy from the grid rather than throwing it away.
2. I use tibber for dynamic power pricing
   → Mostly, if PV power is higher, the price is also low. If I switch on a large consumer (like charging the car), I'd really prefer to use all the power from the grid and save the stored energy for later, when prices will be high again. This results in additional code for the "unimeter quirks", as I use uni-meter to prevent discharging in that scenario.
3. My source sensor is the tibber pulse, which is read locally in my home network. Sometimes new values come in very fast, sometimes it takes 6 or 8 or 40s for a new value to arrive. I want to detect that and steer the output towards 0 in that case, as any other value makes no sense. I therefore create the dummy sensor that updates every second and call it in the main template.

Anticipating that other people will have different needs, I decided to make the priorities easily customizable: Think about what you hate the most: feeding into the grid or drawing power from the grid, then set the constants HATE_FEED_IN and HATE_RECEIVE accordingly. Personally, I have 70% hate for throwing away and 30% for receiving, so I will tweak the output towards receving if it is impossible to stay at zero.