Implementation without TCP-Modbus (incl. demo code)

[m0rph3usX]

Is your feature request related to a problem? Please describe. Currently you have to enable TCP Modbus, which can only be activated with technical-login

Describe the solution you'd like It is possible to fetch all data that are needed without TCP Modbus, by using python and a simple ethernet connection.

Describe alternatives you've considered I debugged the web-interface of IDM: The only thing the python script needs, is the ip-address and the password for login. The password can be set at display panel on the idm heatpump (without technical-login). The cool thing is, that via web-login, automatically all "heating circuits" / "Heizkreise" will be detected.

Here is a demo, which isn't working on my home-assistent yet, but the python script is working and gets all data which are needed. The trick is to extract the " csrf_token" on website login:

import logging
from datetime import timedelta
import requests

from homeassistant.components.sensor import PLATFORM_SCHEMA
import homeassistant.helpers.config_validation as cv
from homeassistant.const import (
    CONF_pin,CONF_ipaddress, CONF_SCAN_INTERVAL, CONF_RESOURCES)
from homeassistant.util import Throttle
from homeassistant.helpers.entity import Entity

_LOGGER = logging.getLogger(__name__)

MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=300)

SENSOR_PREFIX = 'Heating '

SENSOR_TYPES = {
    'mode': ['mode', '', 'mdi:settings'],
    'circuit_mode': ['circuit mode', '', 'mdi:settings'],
    'errors': ['errors', '#', 'mdi:alert-circle'],
    'heat_quantity': ['heat quantity', 'kWh', 'mdi:radiator'],
    'outside_temp': ['outside temp.', '°C', 'mdi:thermometer'],
    'forerun_temp_actual': ['forerun temp. (actual)', '°C', 'mdi:temperature-celsius'],
    'forerun_temp_target': ['forerun temp. (target)', '°C', 'mdi:temperature-celsius'],
    'room_temp_target': ['room temp. (target)', '°C', 'mdi:thermometer'],
    'return_flow_temp': ['return flow temp.', '°C', 'mdi:temperature-celsius'],
    'hygienic_temp': ['hygienic temp.', '°C', 'mdi:temperature-celsius'],
    'water_temp': ['water temp.', '°C', 'mdi:temperature-celsius'],
}

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_PIN): cv.string,
    vol.Required(CONF_IPADDRESS): cv.string,
    vol.Required(CONF_RESOURCES, default=[]):
    vol.All(cv.ensure_list, [vol.In(SENSOR_TYPES)]),
})

def setup_platform(hass, config, add_entities, discovery_info=None):
    _LOGGER.debug("Setup IDM Terra sensors")

    # ipaddress    = config.get(CONF_IPADDRESS)
    # pin          = config.get(CONF_PIN)
    # scanInterval = config.get(CONF_SCAN_INTERVAL)

    ipaddress = "192.168.178.109"
    pin       = "666"

    try:
        data = IdmData(ipaddress, pin)
    except requests.exceptions.HTTPError as error:
        _LOGGER.error(error)
        return False

    entities = []

    for resource in config[CONF_RESOURCES]:
        sensor_type = resource.lower()

        if sensor_type not in SENSOR_TYPES:
            SENSOR_TYPES[sensor_type] = [
                sensor_type.title(), '', 'mdi:flash']

        entities.append(IdmHeatingSensor(data, sensor_type))

    add_entities(entities)

class IdmData(object):
    def __init__(self, ipaddress, pin):
        self._ipaddress = ipaddress
        self._pin       = pin

        self.info              = None
        self.heatpump          = None
        self.pv                = None
        self.stat_baenergyhp   = None
        self.stat_bapv         = None
        self.stat_heatpump     = None
        self.stat_amountofheat = None

    @Throttle(MIN_TIME_BETWEEN_UPDATES)
    def update(self):
        _LOGGER.debug("updating IDM data")

        headers = {
            "Connection": "Keep-Alive",
        }
        params = {
            "pin": self.pin
        }

        self.info              = None
        self.heatpump          = None
        self.pv                = None
        self.stat_baenergyhp   = None
        self.stat_bapv         = None
        self.stat_heatpump     = None
        self.stat_amountofheat = None

        try:
            with requests.Session() as s:
                # get cookie
                r = s.post("http://" + self._ipaddress, headers=headers)

                # enter pin
                r = s.post("http://" + self._ipaddress +"/index.php", headers=headers, data=params)

                #extract csrf_token
                keyname = "csrf_token="
                txt     = r.text
                start = txt.find(keyname) + len(keyname) + 1 # get start of csrf_token entry

                j=start;

                csrf_token =""
                while(j < len(txt)):
                    csrf_token = csrf_token + txt[j]
                    j = j + 1

                    if(txt[j] == '"'):
                        break

                headers["CSRF-Token"] = csrf_token

                #read data
                self.info              = s.get("http://" + self._ipaddress +"/data/info.php", headers=headers).json()
                self.heatpump          = s.get("http://" + self._ipaddress +"/data/heatpump.php", headers=headers).json()
                self.stat_baenergyhp   = s.get("http://" + self._ipaddress +"/data/statistics.php?type=baenergyhp", headers=headers).json()
                self.stat_bapv         = s.get("http://" + self._ipaddress +"/data/statistics.php?type=bapv", headers=headers).json()
                self.stat_heatpump     = s.get("http://" + self._ipaddress +"/data/statistics.php?type=heatpump", headers=headers).json()
                self.stat_amountofheat = s.get("http://" + self._ipaddress +"/data/statistics.php?type=amountofheat", headers=headers).json()    

                return True
        except requests.exceptions.RequestException as exc:
            _LOGGER.error("Error occurred while fetching data: %r", exc)
            self.data = None
            return False

class IdmHeatingSensor(Entity):

    def __init__(self, data, sensor_type):
        self.data   = data
        self.type   = sensor_type
        self._name  = SENSOR_PREFIX + SENSOR_TYPES[self.type][0]
        self._unit  = SENSOR_TYPES[self.type][1]
        self._icon  = SENSOR_TYPES[self.type][2]
        self._state = None

    @property
    def name(self):
        return self._name

    @property
    def icon(self):
        return self._icon

    @property
    def state(self):
        return self._state

    @property
    def unit_of_measurement(self):
        return self._unit

    def update(self):
        self.data.update()
        heatingData = self.data.data

        try:

# self.info              = None
# self.heatpump          = None
# self.pv                = None
# self.stat_baenergyhp   = None
# self.stat_bapv         = None
# self.stat_heatpump     = None
# self.stat_amountofheat = None

            if self.type == 'mode':
                self._state = heatpump["system"]["srcmode"]

            elif self.type == 'circuit_mode':
                for key, value in circuitModeDictionary.items():
                    heatingData["circuits"][0]["mode"] = heatingData["circuits"][0]["mode"].replace(key, value)
                self._state = heatingData["circuits"][0]["mode"]

            elif self.type == 'errors':
                self._state = int(heatingData["error"])

            elif self.type == 'heat_quantity':
                self._state = float(heatingData["sum_heat"].rstrip(" kWh"))

            elif self.type == 'outside_temp':                
                self._state = float(info["out"])

            elif self.type == 'forerun_temp_actual':
                self._state = float(heatingData["circuits"][0]["temp_forerun_actual"].rstrip(" °C"))

            elif self.type == 'forerun_temp_target':
                self._state = float(heatingData["circuits"][0]["temp_forerun"].rstrip(" °C"))

            elif self.type == 'room_temp_target':
                self._state = float(heatingData["circuits"][0]["temp_room_value"])

            elif self.type == 'return_flow_temp':
                self._state = float(heatingData["temp_heat"].rstrip(" °C"))

            elif self.type == 'hygienic_temp':
                self._state = float(heatingData["temp_hygienic"].rstrip(" °C"))

            elif self.type == 'water_temp':
                self._state = float(heatingData["temp_water"].rstrip(" °C"))
        except ValueError:
            self._state = None

A clear and concise description of any alternative solutions or features you've considered.

Additional context Add any other context or screenshots about the feature request here.

[kodebach]

I know it would technically be possible to scrape the data from the web interface via HTTP requests. However, there are multiple reasons why I didn't implement it this way.

The main reason is that the HTTP API you're using is not a public API AFAIK and could change in arbitrary and incompatible ways with any update IDM deploys. My understanding is that this API is designed purely for the web interface, so IMO it's not unlikely that a small change in the web interface could also cause a change in the API.

The Modbus API on the other hand is explicitly meant for communication with external systems and will remain compatible in future versions. AFAIK Modbus is also used by some major 3rd party integrations like the integration of Fronius solar inverters, so it is unlikely IDM would introduce a breaking change.

Another reason is that I didn't want to put in the work of reverse engineering the HTTP API, since the Modbus API is well documented by IDM.

In short, using the HTTP API is possible yes, but out of scope for this integration. If you do not want to enable Modbus for whatever reason, you'll have to write your own separate integration. If you want to do that, you may want to take a look at the Home Assistant Developer Docs and the HACS Docs. In particular you should probably look into the DataUpdateCoordinator, since the heat pump provides quite a lot of sensors which can be fetch together. You may also wanna look at httpx for an HTTP library with async support.