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: "<...>"