Dinotec Pool dosing system

Hello,
I have a Dinotec dosing system for my pool. The PH value, the chlorine value and the temperature are regulated. The system has WiFi and the data is sent to the Dinotec Cloud. I can access the data using the mobile phone app. Unfortunately I can’t find a Homeassistant addon. Only 3 values ​​would have to be transferred to Homeassistant.

System: Dinotec CF Control 100.

Web

Flyer

What options do I have to get the values ​​in Homeassistant or can someone help me?

I have an Android APK file that I can deploy

Thank you very much, regards Stefan

Hey Stefan,

I have the same requirement. Any luck with your quest sofar?

regards,

Theo

Hello, I haven’t had any luck yet.
How difficult is it to develop an application that logs into a cloud and queries 3 values. Regards Stefan

Hello,

Any chance there is an update for this? I’m also very much interested in adding my Dinotec Pool dosing system to home assistant. I did a little bit of digging (unfortunatelly this is way out of my comfort zone) and found that the data is basically retrieved by an API call that looks like this:

wss://remote.dinotec.de:3006/socket.io/?auth_token=eyJ0eXAiOiJ[...]ok&EIO=3&transport=websocket&sid=gjJ2[...]

Response will look like this:
{"7":231,"8":1200,"38":7,"39":1,"40":16,"41":43,"48":0,"49":1,"66":702,"67":718,"68":2048,"69":0,"85":0,"86":587,"87":11,"o":1,"RSSI":-86,"device":"onhEPSHfqJ7kWswy",o:1}

67 is most likely pH value, 66 Redox.
40 and 41 might be the system time.
85 probe voltage, 86 probe slope

So it feels like there is a chance to get these values and for an experienced developer it might be pretty easy. Does anybody know how to set this up?

Sascha

I think I got it to work. It took a long debate with ChatGPT, but here are the final cleaned up instructions:

:person_swimming: Dinotec to Home Assistant via WebSocket and AppDaemon (Full Guide)


:package: 1. Install the AppDaemon Add-on

  1. In Home Assistant:
  • Go to Settings > Add-ons > Add-on Store
  • Search for AppDaemon and install it
  1. In the AppDaemon settings:
  • Enable Start on boot
  • Enable Show in sidebar (optional)
  1. Start the AppDaemon add-on
  2. Open the Log tab and confirm you see:
INFO AppDaemon: You are now ready to run Apps!

:magnifying_glass_tilted_left: 2. Locate the apps Folder

The AppDaemon apps folder is located at:

/config/appdaemon/apps

:paperclip: Tip: If it does not exist, see this helpful thread.


:snake: 3. Install python-socketio (Client Library)

You must add the dependency in AppDaemon’s config to make WebSocket connections.

  1. Go to Settings > Add-ons > AppDaemon > Configuration
  2. Update it to include:
system_packages: []
python_packages:
  - "python-socketio[client]"
init_commands: []
  1. Click Save
  2. Go to the Info tab and Restart AppDaemon

This installs python-socketio into the AppDaemon container.


:brain: 4. Create the Dinotec WebSocket Script

  1. In /config/appdaemon/apps/, create a file: dinotec.py
  2. Paste the following:
import appdaemon.plugins.hass.hassapi as hass
import socketio
import logging

class DinotecClient(hass.Hass):

    def initialize(self):
        # Enable debug logging for troubleshooting
        logging.getLogger('socketio').setLevel(logging.DEBUG)
        logging.getLogger('engineio').setLevel(logging.DEBUG)

        self.log("Starting Dinotec WebSocket Client")
        self.sio = socketio.Client()

        # Handle successful connection
        @self.sio.event
        def connect():
            self.log("Connected to Dinotec")
            client_id = "onhEPSHfqJ7kWswy"  # Replace if dynamic
            self.sio.emit("subscribe-dinoaccess", client_id)
            self.log(f"Sent subscribe-dinoaccess event with client_id: {client_id}")

        # Handle disconnection
        @self.sio.event
        def disconnect():
            self.log("Disconnected from Dinotec")

        # Handle the incoming data and create sensors
        @self.sio.on("dinoaccess-status")
        def handle_dinoaccess(data):
            self.log(f"Received dinoaccess-status: {data}")

            # Mapping from data keys to human-readable sensor names
            sensor_map = {
                "7": ("Pool Temperature", 0.1, "°C"),
                "66": ("Pool Redox", None, "mV"),
                "67": ("Pool pH", 0.01, None),
                "85": ("Pool pH Probe Voltage", None, "mV"),
                "86": ("Pool pH Probe Slope", None, None),
                "87": ("Pool pH Probe Zero Point", None, "mV"),
            }

            for key, value in data.items():
                if str(key) in sensor_map:
                    friendly_name, multiplier, unit = sensor_map[key]
                    entity_id = "sensor." + friendly_name.replace(" ", "_").lower()
                    self.log(entity_id)
                    attributes = {"friendly_name": friendly_name}
                    if unit is not None:
                        attributes["unit_of_measurement"] = unit
                    if multiplier is not None:
                        value *= multiplier

                    self.set_state(entity_id, state=value, attributes=attributes)

        # Attempt to connect to the WebSocket server
        try:
            self.sio.connect(
                f'https://remote.dinotec.de:3006/socket.io/?auth_token={self.args["auth_token"]}&EIO=3',
                transports=['websocket']
            )
        except Exception as e:
            self.log(f"Connection failed: {e}")

:gear: 5. Create or Update apps.yaml

  1. In the same folder (/config/appdaemon/apps/), create a file (or update): apps.yaml
  2. Paste this, replacing the token:
dinotec:
  module: dinotec
  class: DinotecClient
  auth_token: "your_valid_token_here"

:light_bulb: The auth_token is a long JWT you copy from your browser dev tools when logged in to Dinotec.

Does this also work for the CF100?

CF Control

I only have an app on my phone I can use, and there is an internal IP address i can go to, which gives me very basic info

I tried to query the CF100, but I didn't manage to do it.

How do I get my client_id = "onhEPSHfqJ7kWswy" # Replace if dynamic?

@casaro , i hope you can help us to integrate the CF100 in HA :slight_smile: @tomspeijer

Via the page https://remote.dinotec.de I can query my system with my username and password. Where do I have to enter this data in the code? I also get a token displayed in the app.

in the webbrowser , i don`t need a token, only the login

67 is most likely pH value, 66 Redox.

always a connection timeout.

the token is in the apps.yaml

dinotec_client:
  module: dinotec_client
  class: DinotecClient
  auth_token: "my own token"
  client_id: "onhEPSHfqJ7kWswy"

here ist the dinotec_client.py

import appdaemon.plugins.hass.hassapi as hass
import socketio
import logging


class DinotecClient(hass.Hass):

    def initialize(self):
        # Debug-Logging aktivieren
        logging.getLogger("socketio").setLevel(logging.DEBUG)
        logging.getLogger("engineio").setLevel(logging.DEBUG)

        self.log("Starting Dinotec WebSocket Client")

        # Werte aus apps.yaml laden
        self.auth_token = self.args.get("auth_token")
        self.client_id = self.args.get("client_id", "onhEPSHfqJ7kWswy")

        if not self.auth_token:
            self.log("ERROR: auth_token fehlt in apps.yaml", level="ERROR")
            return

        # Socket.IO Client erstellen
        self.sio = socketio.Client(
            logger=True,
            engineio_logger=True,
            reconnection=True,
            reconnection_attempts=0,
            reconnection_delay=5
        )

        # -------------------------
        # Events
        # -------------------------

        @self.sio.event
        def connect():
            self.log("Connected to Dinotec")
            self.sio.emit("subscribe-dinoaccess", self.client_id)
            self.log(f"Sent subscribe-dinoaccess with client_id: {self.client_id}")

        @self.sio.event
        def disconnect():
            self.log("Disconnected from Dinotec", level="WARNING")

        @self.sio.event
        def connect_error(data):
            self.log(f"Connection error: {data}", level="ERROR")

        @self.sio.on("dinoaccess-status")
        def handle_dinoaccess(data):
            self.log(f"Received dinoaccess-status: {data}")

            sensor_map = {
                "7": ("Pool Temperature", 0.1, "°C"),
                "66": ("Pool Redox", None, "mV"),
                "67": ("Pool pH", 0.01, None),
                "85": ("Pool pH Probe Voltage", None, "mV"),
                "86": ("Pool pH Probe Slope", None, None),
                "87": ("Pool pH Probe Zero Point", None, "mV"),
            }

            if not isinstance(data, dict):
                self.log(f"Unexpected data format: {data}", level="WARNING")
                return

            for key, value in data.items():
                key_str = str(key)

                if key_str not in sensor_map:
                    continue

                friendly_name, multiplier, unit = sensor_map[key_str]
                entity_id = "sensor." + friendly_name.replace(" ", "_").lower()

                try:
                    if value is not None:
                        value = float(value)
                        if multiplier is not None:
                            value = value * multiplier
                except Exception as e:
                    self.log(
                        f"Could not convert value for key {key_str}: {value} ({e})",
                        level="WARNING"
                    )
                    continue

                attributes = {
                    "friendly_name": friendly_name
                }

                if unit is not None:
                    attributes["unit_of_measurement"] = unit

                self.set_state(entity_id, state=value, attributes=attributes)
                self.log(f"Updated {entity_id} = {value}")

        # -------------------------
        # Verbindung aufbauen
        # -------------------------
        url = f'https://remote.dinotec.de:3006/socket.io/?auth_token={self.auth_token}&EIO=3'

        try:
            self.log(f"Connecting to {url}")
            self.sio.connect(
                url,
                transports=["websocket"],
                wait_timeout=20
            )
            self.log("Socket.IO connection started")
        except Exception as e:
            self.log(f"Connection failed: {e}", level="ERROR")```

I got it to work.
For the CF100

There was a major refactor of the code on the DinoTec side. My current implementation (which works fine) looks like this:

import appdaemon.plugins.hass.hassapi as hass
import requests
import re
import json
import html

# Mapping from data keys to human-readable sensor names
SENSOR_MAP = {
  "7": ("Pool Temperature", 0.1, "°C"),
  "8": ("Pool Temperature Extern", 0.1, "°C"),
  "39": ("Pool Unknown Sensor", None, None),
  "66": ("Pool Redox", None, "mV"),
  "67": ("Pool pH", 0.01, "pH"),
  "85": ("Pool pH Probe Voltage", None, "mV"),
  "86": ("Pool pH Probe Slope", None, "mV/pH"),
  "87": ("Pool pH Probe Zero Point", None, "mV"),
  "RSSI": ("Pool Signal Strength", None, "dBm"),
}

UNAVAILABLE = "unavailable"
UNKNOWN = "unknown"

# Constants
BASE_URL = "https://remote.dinotec.de"
LOGIN_URL = f"{BASE_URL}/login"
AUTH_URL = f"{BASE_URL}/auth/login"

class DinotecClient(hass.Hass):

  def initialize(self):
    self.log("Starting Dinotec System (Production HTTP Scraper Mode)")
    
    self.session = requests.Session()
    self.session.headers.update({
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8'
    })
    
    self.pool_url = None
    
    # Initialize all sensors in Home Assistant immediately so they aren't blank
    for key, sensor_definition in SENSOR_MAP.items():
        entity_id = "sensor." + sensor_definition[0].replace(" ", "_").lower()
        if self.get_state(entity_id) is None:
            self.update_ha(sensor_definition, UNKNOWN)

    # Run the fetch_data function immediately, and then strictly every 30 seconds
    self.run_every(self.fetch_data, "now", 30)

  def do_login(self):
    """Handles session creation, CSRF token extraction, and authentication."""
    email = self.args.get("email")
    password = self.args.get("password")
    
    if not email or not password:
        self.log("[CONFIG_ERROR] Email or Password missing in apps.yaml", level="ERROR")
        return False
        
    try:
        self.log("Session missing or expired. Initiating new login...")
        
        # 1. Fetch Login Page
        res = self.session.get(LOGIN_URL, timeout=15)
        res.raise_for_status()
        
        # 2. Extract CSRF Token
        csrf_match = re.search(r'name="csrf-token"\s+content="([^"]+)"', res.text)
        if not csrf_match:
            self.log("[PARSE_ERROR] Login HTML changed. Could not find 'csrf-token' meta tag.", level="ERROR")
            return False
            
        # 3. Submit Credentials
        payload = {
            "_token": csrf_match.group(1), 
            "email": email, 
            "password": password, 
            "remember": "on"
        }
        res_post = self.session.post(AUTH_URL, data=payload, allow_redirects=True, timeout=15)
        res_post.raise_for_status()
        
        # 4. Extract Installation/Pool ID from the redirect or dashboard HTML
        pool_match = re.search(r'/installation/(\d+)/', res_post.text)
        
        if pool_match:
            pool_id = pool_match.group(1)
            self.pool_url = f"{BASE_URL}/installation/{pool_id}/overview"
            self.log(f"[AUTH_SUCCESS] Logged in successfully. Target Pool Dashboard: {self.pool_url}")
            return True
        else:
            # If the ID isn't found, check if we were rejected back to the login page
            if "/login" in res_post.url:
                self.log("[AUTH_ERROR] Server rejected credentials. Check email/password.", level="ERROR")
            else:
                self.log("[PARSE_ERROR] Logged in, but could not find the 'Installation ID' regex in dashboard HTML.", level="ERROR")
            return False
            
    except requests.exceptions.RequestException as e:
        self.log(f"[NETWORK_ERROR] Failed during login request: {e}", level="ERROR")
        return False
    except Exception as e:
        self.log(f"[SYSTEM_ERROR] Unexpected error during login: {e}", level="ERROR")
        return False

  def fetch_data(self, kwargs):
    """Core loop: Fetches the dashboard and parses the embedded JSON data."""
    # Ensure we are logged in before fetching
    if not self.pool_url:
        if not self.do_login():
            return # Will automatically retry in 30 seconds on next cycle
            
    try:
        # Fetch the Pool Overview page
        res = self.session.get(self.pool_url, timeout=15)
        res.raise_for_status()
        
        # Detect session expiration (Server redirects us to login)
        if "/login" in res.url:
            self.log("[AUTH_EXPIRED] Session cookie expired. Forcing re-login on next cycle.", level="WARNING")
            self.pool_url = None
            self.session.cookies.clear()
            return
            
        # Extract the JSON payload hidden in the HTML data attribute
        data_match = re.search(r'data-recent-registers=[\'"](.*?)[\'"]', res.text, re.S)
        
        if data_match:
            raw_json = html.unescape(data_match.group(1))
            data = json.loads(raw_json)
            
            # Check pump state to determine if sensors should be marked unavailable
            pump_state = self.get_state("switch.poolpumpe")
            
            # Update sensors
            updated_count = 0
            for key, sensor_def in SENSOR_MAP.items():
                if key in data:
                    self.update_ha(sensor_def, data[key])
                    updated_count += 1
                else:
                    if pump_state == "on" or pump_state is None:
                        self.update_ha(sensor_def, UNAVAILABLE)
            
            # Use DEBUG to avoid spamming the log every 30 seconds. 
            # (Change to self.log if you want to see it continuously).
            self.log(f"Cycle complete. Updated {updated_count} sensors.", level="DEBUG")
            
        else:
            self.log("[PARSE_ERROR] 'data-recent-registers' JSON not found on overview page. Did Dinotec update their layout?", level="ERROR")
            
    except requests.exceptions.RequestException as e:
        self.log(f"[NETWORK_ERROR] Timeout or connection failure during data fetch: {e}", level="WARNING")
    except json.JSONDecodeError as e:
        self.log(f"[PARSE_ERROR] Found data string, but it is not valid JSON: {e}", level="ERROR")
    except Exception as e:
        self.log(f"[SYSTEM_ERROR] Unexpected error during fetch: {e}", level="ERROR")

  def update_ha(self, sensor_definition, value):
    """Helper method to format values and push them to Home Assistant."""
    friendly_name, multiplier, unit = sensor_definition
    entity_id = "sensor." + friendly_name.replace(" ", "_").lower()
    
    attributes = {"friendly_name": friendly_name}
    if unit:
        attributes["unit_of_measurement"] = unit
    
    if multiplier is not None and value not in [UNAVAILABLE, UNKNOWN]:
        try:
            value = round(multiplier * float(value), 5)
        except (ValueError, TypeError):
            value = UNAVAILABLE
            
    self.set_state(entity_id, state=value, attributes=attributes)

The apps.yaml file looks like this. So you don’t need an auth token anymore.

dinotec:
  module: dinotec
  class: DinotecClient
  email: "<...>"
  password: "<...>"

I used another approach.
Have a python script running on my vm and it’s sending data towards 2 MQTT sensors