Skip to content

Helper API

Jump to bottom
Jeroen Simonetti edited this page Apr 9, 2026 · 1 revision

Developer Guide: Building a Custom Helper

This page documents the MQTT API contract that mimirheim expects from input helpers. Follow this contract to build a helper that feeds any data source into mimirheim, regardless of whether it involves an API, a database, a local sensor, or a custom algorithm.

What is a helper?

A helper is a standalone daemon that:

  1. Obtains data from some source (web API, database, Home Assistant, a flat file, etc.)
  2. Formats that data as a mimirheim-compatible JSON payload
  3. Publishes the payload to the correct MQTT topic (retained)
  4. Optionally publishes an empty message to {prefix}/input/trigger to make mimirheim solve immediately

That is the complete contract. A helper does not need to know anything about the solver, the config schema, or the schedule format. It only needs to know the MQTT topic and payload format for the data it provides.

Available input topics

mimirheim subscribes to a fixed set of input topic patterns. Which topics are actually watched depends on the config.yaml (specifically, which device types are configured and whether inputs are enabled). The table below lists every publishable topic type and its payload format.

All topics are retained unless otherwise noted. Retained means the MQTT broker stores the last message and replays it to any new subscriber, including mimirheim on startup.

Topic: Prices

Topic: {prefix}/input/prices

Format: JSON array of hourly price steps

[
  {
    "ts": "2026-03-30T13:00:00+00:00",
    "import_eur_per_kwh": 0.2418,
    "export_eur_per_kwh": 0.1952,
    "confidence": 1.0
  },
  {
    "ts": "2026-03-30T14:00:00+00:00",
    "import_eur_per_kwh": 0.2112,
    "export_eur_per_kwh": 0.1847
  }
]

Rules:

  • ts is an ISO 8601 UTC datetime string marking the start of that price period.
  • import_eur_per_kwh and export_eur_per_kwh are in EUR/kWh. All-in prices including taxes.
  • confidence is optional; defaults to 1.0 when absent. Use 1.0 for guaranteed day-ahead prices. Use values below 1.0 for forecast prices.
  • Steps may be at any resolution (hourly is standard for day-ahead markets). mimirheim uses a step function: the price for ts applies until the next ts in the array.
  • The planning horizon ends at the last ts in the array. The array must extend far enough ahead to cover readiness.min_horizon_hours.

Topic: PV generation forecast

Topic: configured as pv_arrays.{name}.topic_forecast
Default derivation: {prefix}/input/pv/{name}/forecast

Format: JSON array of power forecast steps

[
  {"ts": "2026-03-30T10:00:00+00:00", "kw": 0.0},
  {"ts": "2026-03-30T11:00:00+00:00", "kw": 1.8, "confidence": 0.90},
  {"ts": "2026-03-30T12:00:00+00:00", "kw": 4.2, "confidence": 0.75}
]

Rules:

  • kw is forecast output power in kilowatts. Must be ≥ 0.
  • confidence is optional; defaults to 1.0.
  • mimirheim resamples to its 15-minute solver grid using linear interpolation between adjacent known points.
  • Publish one topic per PV array. The device name in the topic must match the key in mimirheim's pv_arrays config section (or the topic must be set explicitly in that config).

Topic: Static load forecast

Topic: configured as static_loads.{name}.topic_forecast
Default derivation: {prefix}/input/baseload/{name}/forecast

Format: same as PV forecast

[
  {"ts": "2026-03-30T10:00:00+00:00", "kw": 0.45},
  {"ts": "2026-03-30T11:00:00+00:00", "kw": 0.42}
]

Rules:

  • kw is forecast demand in kilowatts. Must be ≥ 0.
  • mimirheim resamples using linear interpolation.
  • One topic per static load device.

Topic: Trigger

Topic: configured as {prefix}/input/trigger
Retained: no (any retained trigger would fire mimirheim indefinitely on reconnect)

After publishing all data for a cycle, publish any message (including an empty payload "") to the trigger topic:

mqtt_client.publish("mimir/input/trigger", payload="", retain=False)

mimirheim checks readiness on every trigger. If all required data is present and the horizon is sufficient, it solves. If not, it logs which inputs are missing or stale and takes no action.

Important: do not publish to the trigger topic from a retained publish. If you restart your helper and the broker replays a retained trigger, mimirheim would solve on potentially stale data.

Topic: Battery SOC

Topic: configured as batteries.{name}.inputs.soc.topic
Default derivation: {prefix}/input/battery/{name}/soc

Format: plain numeric string

72.5

The unit is determined by batteries.{name}.inputs.soc.unit in mimirheim's config:

  • unit: percent — the value is a 0–100 percentage. mimirheim converts to kWh using capacity_kwh.
  • unit: kwh — the value is an absolute energy in kWh.

Topic: EV charger state

Topic: configured as ev_chargers.{name}.inputs.soc.topic
Default derivation: {prefix}/input/ev/{name}/soc

Format: plain numeric string (SOC value in configured unit), or JSON with optional departure target:

{
  "soc": 72.5,
  "target_soc_kwh": 48.0,
  "window_latest": "2026-03-30T08:00:00+00:00"
}
Field Type Required Description
soc float Yes (if JSON) Current state of charge in the configured unit
target_soc_kwh float No Required SOC at departure. Hard constraint in solver.
window_latest ISO 8601 UTC string No (if no target) Departure deadline for the target. Required when target_soc_kwh is set.

A plain numeric string is treated as the SOC value with no target or departure window.

Plug state topic:

Topic: configured as ev_chargers.{name}.inputs.plugged_in_topic
Default derivation: {prefix}/input/ev/{name}/plugged_in

Format: boolean-like string: true/false, on/off, or 1/0 (case-insensitive).

When the plug state is false (or the topic has not been received), the EV device is excluded from the model entirely.

Topic: Hybrid inverter SOC

Topic: configured as hybrid_inverters.{name}.inputs.soc.topic
Default derivation: {prefix}/input/hybrid/{name}/soc

Same format as battery SOC (plain numeric string, unit from config).

Topic: Hybrid inverter PV forecast

Topic: configured as hybrid_inverters.{name}.topic_pv_forecast
Default derivation: {prefix}/input/hybrid/{name}/pv_forecast

Same format as PV array forecast.

Topic: Thermal boiler temperature

Topic: configured as thermal_boilers.{name}.inputs.topic_current_temp
Default derivation: {prefix}/input/thermal_boiler/{name}/temp_c

Format: plain numeric string (degrees Celsius)

58.3

Topic: Space heating demand

Topic: configured as space_heating_hps.{name}.inputs.topic_heat_needed_kwh
Default derivation: {prefix}/input/space_heating/{name}/heat_needed_kwh

Format: plain numeric string (kWh — total heat demand over the planning horizon)

12.4

Not required when building_thermal is configured on the heat pump.

Topic: BTM indoor temperature

Topic: configured as *.building_thermal.inputs.topic_current_indoor_temp_c
Default derivation: {prefix}/input/[space_heating|combi_hp]/{name}/btm/indoor_temp_c

Format: plain numeric string, or JSON object:

20.5

or

{"temp_c": 20.5}

Topic: BTM outdoor temperature forecast

Topic: configured as *.building_thermal.inputs.topic_outdoor_temp_forecast_c
Default derivation: {prefix}/input/[space_heating|combi_hp]/{name}/btm/outdoor_forecast_c

Format: JSON array of floats (°C), one value per 15-minute step, starting from the current solve time

[5.2, 5.0, 4.8, 4.7, 4.5, 4.3, 4.1, 4.0]

The array must contain at least as many values as the active horizon length.

Building a helper in Python

Here is a minimal implementation template. Adapt it to your data source.

"""my_helper — example mimirheim input helper.

Fetches data from some source and publishes it to the mimirheim MQTT input topic.
"""

import json
import logging
import sys
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
import yaml

logger = logging.getLogger("my_helper")


def fetch_data() -> list[dict]:
    """Fetch data and return a list of {'ts': ..., 'kw': ..., 'confidence': ...} dicts."""
    raise NotImplementedError("Implement this for your data source")


def run(config: dict) -> None:
    """Connect to MQTT, subscribe to the trigger topic, and handle trigger messages."""
    broker = config["mqtt"]["host"]
    port = config["mqtt"]["port"]
    client_id = config["mqtt"]["client_id"]
    trigger_topic = config["trigger_topic"]
    output_topic = config["output_topic"]
    hioo_trigger = config.get("mimir_trigger_topic")

    def on_connect(client, userdata, flags, rc, properties=None):
        if rc == 0:
            logger.info("connected to broker")
            client.subscribe(trigger_topic)
        else:
            logger.error("connection failed: rc=%s", rc)

    def on_message(client, userdata, msg):
        logger.info("trigger received on %s", msg.topic)
        try:
            steps = fetch_data()
        except Exception:
            logger.exception("fetch failed")
            return

        payload = json.dumps(steps)
        client.publish(output_topic, payload=payload, retain=True)
        logger.info("published %d steps to %s", len(steps), output_topic)

        if hioo_trigger:
            client.publish(hioo_trigger, payload="", retain=False)
            logger.info("signalled mimirheim at %s", hioo_trigger)

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    client.on_connect = on_connect
    client.on_message = on_message

    username = config["mqtt"].get("username")
    password = config["mqtt"].get("password")
    if username:
        client.username_pw_set(username, password)

    client.connect(broker, port)
    client.loop_forever()


def main() -> None:
    with open(sys.argv[1]) as f:
        config = yaml.safe_load(f)
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(name)s: %(message)s",
    )
    run(config)


if __name__ == "__main__":
    main()

Payload validation helpers

The helper_common package (shared by all built-in helpers) provides validated Pydantic models for the MQTT connection (MqttConfig) and HA discovery (HomeAssistantConfig). You can import these in your own helper if you install helper_common as a dependency:

from helper_common.config import MqttConfig, HomeAssistantConfig
import helper_common.topics as _topics

# Get the canonical derived topic for a given prefix and device name:
topic = _topics.pv_forecast_topic("mimir", "my_array")
# → "mimir/input/pv/my_array/forecast"

Available topic derivation helpers in helper_common.topics

Function Returns
prices_topic(prefix) {prefix}/input/prices
trigger_topic(prefix) {prefix}/input/trigger
pv_forecast_topic(prefix, array_name) {prefix}/input/pv/{array_name}/forecast
baseload_forecast_topic(prefix, load_name) {prefix}/input/baseload/{load_name}/forecast
dump_available_topic(prefix) {prefix}/status/dump_available

Readiness and staleness

mimirheim checks readiness before every solve. A retained message that was published once and never updated will eventually become "stale" from mimirheim's perspective: the forecast steps it contains will all have timestamps in the past, leaving zero horizon steps available.

Best practice: re-publish every forecast on each trigger cycle, not just when data changes. Retained MQTT messages do not expire; the staleness is determined by the ts values in the payload, not by the MQTT broker's retention time. If your helper is triggered every hour, the payload should always contain at least 24–48 hours of forecast steps starting from the current hour.

Testing your helper

Use simplebroker.py (included in the scripts/ directory) to run an in-process MQTT broker during development:

python3 scripts/simplebroker.py

Then start your helper pointing at localhost:1883 and subscribe to its output topic with mosquitto_sub or an MQTT client:

mosquitto_sub -h localhost -t "mimir/#" -v

Trigger the helper by publishing to its trigger topic:

mosquitto_pub -h localhost -t my_helper/trigger -m ""

Verify the payload format, then check that mimirheim reads it correctly by watching mimir/status/last_solve.

Home

Getting started

Helpers

Developer

Reference

Clone this wiki locally