Viessman Aquastilla smart Topic is solved

[Mozekam]

Hello everyone,
First of all, I'd like to praise this forum – it's a never-ending source of knowledge and help! I have a quick question. Is there any way to retrieve data from the Viessman Aquastilla app? I know the ViCare app works with Domoticz, but unfortunately, my water softener can only connect to the dedicated Aquastilla app.

Has anyone had any experience with this?

Best regards!

[waltervl]

There is a python library for this. If it works for you it can be used to integrate it as a Python plugin or use it as a simple interface with virtual devices.
https://github.com/alakdae/aquastilla-softener

[Mozekam]

Thank you for your quick reply. Do you have any instructions for someone who's never done this before?

[waltervl]

Invest some time in learning the principles of Python.
Install the python library so you can use it. Instruction should be in the GitHub repository.
Then copy/paste the script example in the readme of the library and create a file test.py.
Modify the contents so it has the correct account credentials.
Then run in a terminal session in the folder where the test.py script is

Code: Select all

python3 test.py

You should see information about the device and some commands are sent to the device (last lines of the script).
When you change the commands you can run the script again.
If you run into errors try chatgpt or copilot or another AI to solve it.

[Mozekam]

Hello, thank you very much for your suggestions - I underestimated the AI ​​tools. Got it sorted. Everything works beautifully.

[waltervl]

Nice!
Can you share your solution? Do you have a github repository?

[Mozekam]

Of course, I will describe everything - I just need a moment, only late yesterday evening I managed to run a satisfactory script

As for github, I don't have it yet.

[Mozekam]

Hello,
Below I'm posting a script for reading data via the Aquastilla API.

It collects data such as:
- salt level
- water used today
- remaining water until the next regeneration
- expected regeneration date
- flood alarm (I have a dedicated wireless sensor connected)
- vacation mode (status preview and remote on/off capability)
- display of the current operating mode

In the future, I'll probably want to add the ability to remotely turn off the water flow – this option is available in the app.

crontab task:

Code: Select all

*/15 * * * * /usr/bin/python3 /home/pi/domoticz/scripts/python/aquastilla_domoticz.py

The script runs once every 15 minutes. I haven't tried running it at a shorter interval, because, according to the information I have, too frequent calls can temporarily lock the account. Secondly, I don't think it makes sense to update such data too frequently.

you need to create a file and paste the following script into it (You'll need to provide your login details, IP address, and optionally, the names of the displayed information—e.g., modes or error messages)

/home/pi/domoticz/scripts/python/aquastilla_domoticz.py

Code: Select all

import requests
import json
from datetime import datetime, timedelta, timezone
from aquastilla_softener import AquastillaSoftener

# =====================================================
# --- KONFIGURACJA DANYCH LOGOWANIA ---
# =====================================================
EMAIL = "TWÓJ_EMAIL"
PASSWORD = "TWOJE_HASLO"
DOMOTICZ_URL = "http://localhost:8080"

# =====================================================
# --- TWOJE NUMERY IDX Z DOMOTICZA ---
# =====================================================
IDX_SOL = 10          
IDX_WODA_DZIS = 11    
IDX_WODA_ZAPAS = 12   
IDX_ALARM = 13        
IDX_DATA_REGEN = 14   
IDX_WAKACJE_SW = 15   
IDX_TRYB = 16         
# =====================================================

def update_domoticz(idx, nvalue, svalue):
    url = f"{DOMOTICZ_URL}/json.htm?type=command&param=udevice&idx={idx}&nvalue={nvalue}&svalue={svalue}"
    try:
        response = requests.get(url, timeout=5)
        return response.status_code == 200
    except Exception as e:
        print(f"Błąd Domoticz (IDX {idx}): {e}")
        return False

# Słownik tłumaczeń trybów (dostosuj, jeśli API zwróci inne nazwy)
TRYBY_PRACY = {
    "SERVICE": "Zmiękczanie",
    "REGENERATION": "Regeneracja",
    "BACKWASH": "Płukanie wsteczne",
    "BRINE": "Solankowanie",
    "RINSE": "Płukanie końcowe",
    "REFILL": "Napełnianie zbiornika",
    "STANDBY": "Oczekiwanie"
}

try:
    softener = AquastillaSoftener(email=EMAIL, password=PASSWORD)
    devices = softener.list_devices()
    
    if not devices:
        print("Nie znaleziono urządzeń.")
    else:
        device = devices[0]
        uuid = device['uuid']
        headers = softener._get_headers()
        base = softener._api_base_url

        res_state = requests.get(f"{base}/device/{uuid}/state", headers=headers)
        data = res_state.json()

        res_settings = requests.get(f"{base}/device/{uuid}/settings", headers=headers)
        settings = res_settings.json()

        # --- OBRÓBKA DANYCH ---
        
        sol = data.get("saltPercent", 0)
        woda_dzis = round(data.get("todayWaterUsage", 0) * 1000, 1)
        woda_zapas = round(data.get("waterLeft", 0) * 1000, 1)
        
        # Pobieranie i mapowanie trybu pracy
        raw_mode = data.get("operationMode", "UNKNOWN").upper()
        tryb_tekst = TRYBY_PRACY.get(raw_mode, f"Inny ({raw_mode})")

        raw_date = data.get("expectedRegenerationDate", "Brak")
        clean_date = raw_date.split("T")[0] if "T" in raw_date else raw_date

        is_vacation = settings.get("vacationMode", False)
        vac_n = 1 if is_vacation else 0
        vac_s = "On" if is_vacation else "Off"

        # Alarm zalania
        flood_status = device.get("deviceHistory", {}).get("flood")
        n_alert, s_alert = 1, "System bezpieczny"
        if flood_status:
            flood_time = datetime.fromisoformat(flood_status.replace("Z", "+00:00"))
            if datetime.now(timezone.utc) - flood_time < timedelta(minutes=30):
                n_alert, s_alert = 4, "ALARM! WYKRYTO ZALANIE!"

        # --- WYSYŁKA DO DOMOTICZA ---
        update_domoticz(IDX_SOL, 0, sol)
        update_domoticz(IDX_WODA_DZIS, 0, woda_dzis)
        update_domoticz(IDX_WODA_ZAPAS, 0, woda_zapas)
        update_domoticz(IDX_DATA_REGEN, 0, clean_date)
        update_domoticz(IDX_ALARM, n_alert, s_alert)
        update_domoticz(IDX_WAKACJE_SW, vac_n, vac_s)
        update_domoticz(IDX_TRYB, 0, tryb_tekst) # <--- Wysyłka trybu

        print(f"[{datetime.now().strftime('%H:%M:%S')}] Aktualizacja OK.")
        print(f"Tryb: {tryb_tekst}, Sól: {sol}%, Woda dziś: {woda_dzis}L")

except Exception as e:
    print(f"Błąd krytyczny: {e}")
    

The attached photo shows what types of virtual sensors should be added in domoticz.

This is my first post, so please be understanding when it comes to descriptions and information

Thanks again to waltervl for effectively pointing me in the right direction

[waltervl]

Thanks for sharing. Next wonderful step would be to make this a python plugin but that is perhaps for another Aquastilla user with python skills to do...

[Mozekam]

Hello everyone.
The Aquastilla API's recalcitrance—frequent lags and server interruptions, plus account lockouts for excessive server requests—has been troubling me. So I decided to investigate the issue a bit. Below is a new script. It turns out the manufacturer (intentionally or unintentionally) left a loophole in the water softener's local API. The most critical parameters, those that should be updated live, can be "stealed" locally. Unfortunately, the control and status of, for example, the flood sensor are still at the mercy of an external server. Since the flood sensor, updated every 30 minutes (which is ultimately the safest time for server requests), is useless, the script monitors the increments of liters of water used over time and displays the current flow rate. If no one is home and a flow is detected, we can remotely shut down the valve.

Code: Select all

import requests
import json
import os
import random
import urllib3
import time
from datetime import datetime

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# --- KONFIGURACJA ---
AQUA_IP = "xxx.xxx.xx.xx"
EMAIL = "[email protected]"
PASSWORD = "xxxxxxxxxxx"
DOMOTICZ_URL = "http://xxx.xxx.xx.xx:8080" 
INFLUX_URL = "http://localhost:8086/write?db=domoticz" 

INTERVAL_FAST = 5.0  
SLOW_MULTIPLIER = 360  
VIRTUAL_LEAK_MINUTES = 20  # <--- Po jakim czasie ciągłego lania wody uruchomić alarm

# --- INDEKSY DOMOTICZ ---
IDX_PRZEPLYW_LIVE = 707   
IDX_WODA_MIESIAC = 587    
IDX_WODA_DZIEN_NOWY = 708  
IDX_WODA_DZIS = 554       
IDX_SOL = 553          
IDX_WODA_ZAPAS = 555   
IDX_ALARM = 556        
IDX_DATA_REGEN = 557   
IDX_TRYB = 559         
IDX_WAKACJE_SW = 558   

# --- PLIKI CACHE ---
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CACHE_TOTAL = os.path.join(BASE_DIR, "woda_suma_total.txt")
CACHE_TODAY_LAST = os.path.join(BASE_DIR, "woda_dzis_punkt_odniesienia.txt")
CACHE_MONTH_REF = os.path.join(BASE_DIR, "ostatni_miesiac.txt")
CACHE_DAILY_TOTAL = os.path.join(BASE_DIR, "woda_dzis_suma.txt")
CACHE_DAY_REF = os.path.join(BASE_DIR, "ostatni_dzien.txt")

try:
    from aquastilla_softener import AquastillaSoftener
except Exception:
    AquastillaSoftener = None

def update_domoticz(session, idx, nvalue, svalue):
    if idx == 587:
        svalue = round(float(svalue) / 1000, 3)
    url = f"{DOMOTICZ_URL}/json.htm?type=command&param=udevice&idx={idx}&nvalue={nvalue}&svalue={str(svalue)}"
    try:
        session.get(url, timeout=1.5)
    except:
        pass

def main():
    print("🚀 Kombajn V11 (Z Wirtualnym Czujnikiem Wycieku) wystartował...")
    session = requests.Session()
    session.verify = False
    session.headers.update({'Connection': 'keep-alive'})
    
    cycle_count = 0
    last_hourly_water = None  
    is_leak = False
    is_vacation = False
    force_slow_run = True
    
    flow_history = [0.0] * 6
    
    # Zmienne dla wirtualnego czujnika zalania
    continuous_flow_seconds = 0
    virtual_leak_triggered = False

    while True:
        start_time = time.time()
        now_dt = datetime.now()
        current_month_str = now_dt.strftime("%Y-%m")
        current_day_str = now_dt.strftime("%Y-%m-%d")
        
        res_local = None
        try:
            res_local = session.get(f"https://{AQUA_IP}/status", timeout=1.8).json()
        except:
            pass

        # === 1. PĘTLA WOLNA (CO 30 MINUT) ===
        if cycle_count >= SLOW_MULTIPLIER or force_slow_run:
            cycle_count = 0
            force_slow_run = False
            
            if res_local:
                raw_timestamp = res_local.get("expectedRegenerationTime", 0)
                formatted_date = datetime.fromtimestamp(raw_timestamp).strftime('%Y-%m-%d') if raw_timestamp > 0 else "Brak"
                update_domoticz(session, IDX_DATA_REGEN, 0, formatted_date)

            if AquastillaSoftener is not None:
                try:
                    softener = AquastillaSoftener(email=EMAIL, password=PASSWORD)
                    device = softener.list_devices()[0]
                    uuid = device['uuid']
                    headers = softener._get_headers()
                    base = softener._api_base_url
                    v = random.randint(1, 10000)
                    
                    res_settings = requests.get(f"{base}/device/{uuid}/settings?v={v}", headers=headers, timeout=8).json()
                    res_state = requests.get(f"{base}/device/{uuid}/state?v={v}", headers=headers, timeout=8).json()
                    
                    raw_mode = res_state.get("state", "UNKNOWN")
                    mode_pl = "Zmiękczanie" if "Softening" in raw_mode else "Regeneracja" if "Regen" in raw_mode else raw_mode
                    is_leak = res_state.get("leakDetected", False)
                    is_vacation = res_settings.get("vacationMode", False)
                    
                    update_domoticz(session, IDX_TRYB, 0, mode_pl)
                    update_domoticz(session, IDX_WAKACJE_SW, (1 if is_vacation else 0), ("On" if is_vacation else "Off"))
                    
                    # Sprzętowy alarm z chmury (nadpisywany przez wirtualny, jeśli wirtualny wyłapie wyciek pierwszy)
                    if not virtual_leak_triggered:
                        status_alarm = "Brak wycieku ✅" if not is_leak else "Wyciek (Czujnik Stacji)! ❗"
                        n_val_alarm = 1 if not is_leak else 4
                        update_domoticz(session, IDX_ALARM, n_val_alarm, status_alarm)
                except:
                    pass

        # === 2. PĘTLA SZYBKA (CO 5 SEKUND) ===
        if res_local:
            try:
                current_hourly_water = float(res_local.get("waterCurrentHour", 0.0)) * 1000
                raw_flow = 0.0
                delta_liters = 0.0
                
                if last_hourly_water is not None:
                    delta_liters = current_hourly_water - last_hourly_water
                    if delta_liters >= 0:
                        raw_flow = delta_liters * (60.0 / INTERVAL_FAST)
                    else:
                        raw_flow = 0.0
                        delta_liters = 0.0
                
                last_hourly_water = current_hourly_water
                
                flow_history.pop(0)
                flow_history.append(raw_flow)
                smoothed_flow = round(sum(flow_history) / len(flow_history), 1)
                if smoothed_flow < 0.2:
                    smoothed_flow = 0.0
                
                update_domoticz(session, IDX_PRZEPLYW_LIVE, 0, smoothed_flow)
                
                # --- WIRTUALNY CZUJNIK ZALANIA ---
                if smoothed_flow > 0.0:
                    continuous_flow_seconds += INTERVAL_FAST
                else:
                    continuous_flow_seconds = 0
                    if virtual_leak_triggered:
                        virtual_leak_triggered = False
                        if not is_leak:
                            update_domoticz(session, IDX_ALARM, 1, "Brak wycieku ✅")

                if continuous_flow_seconds >= (VIRTUAL_LEAK_MINUTES * 60) and not virtual_leak_triggered:
                    virtual_leak_triggered = True
                    update_domoticz(session, IDX_ALARM, 4, f"Ciągły pobór (>{VIRTUAL_LEAK_MINUTES}m)! ❗")

                # --- ODDZIELNA LOGIKA RESETU DOBOWEGO ---
                if os.path.exists(CACHE_DAY_REF):
                    with open(CACHE_DAY_REF, "r") as f: last_saved_day = f.read().strip()
                else:
                    last_saved_day = current_day_str
                    with open(CACHE_DAY_REF, "w") as f: f.write(current_day_str)

                if not os.path.exists(CACHE_DAILY_TOTAL):
                    current_daily_total = 0.0
                    with open(CACHE_DAILY_TOTAL, "w") as f: f.write("0.0")
                else:
                    with open(CACHE_DAILY_TOTAL, "r") as f:
                        try: current_daily_total = float(f.read())
                        except: current_daily_total = 0.0

                if last_saved_day != current_day_str:
                    current_daily_total = 0.0
                    with open(CACHE_DAY_REF, "w") as f: f.write(current_day_str)
                    with open(CACHE_DAILY_TOTAL, "w") as f: f.write("0.0")

                # --- ODDZIELNA LOGIKA RESETU MIESIĘCZNEGO ---
                if os.path.exists(CACHE_MONTH_REF):
                    with open(CACHE_MONTH_REF, "r") as f: last_saved_month = f.read().strip()
                else:
                    last_saved_month = current_month_str
                    with open(CACHE_MONTH_REF, "w") as f: f.write(current_month_str)

                if not os.path.exists(CACHE_TOTAL):
                    current_total = 0.0
                    with open(CACHE_TOTAL, "w") as f: f.write("0.0")
                else:
                    with open(CACHE_TOTAL, "r") as f:
                        try: current_total = float(f.read())
                        except: current_total = 0.0

                if last_saved_month != current_month_str:
                    current_total = 0.0
                    with open(CACHE_MONTH_REF, "w") as f: f.write(current_month_str)
                    with open(CACHE_TOTAL, "w") as f: f.write("0.0")

                # --- AKTUALIZACJA LICZNIKÓW O DELTĘ ---
                if os.path.exists(CACHE_TODAY_LAST):
                    with open(CACHE_TODAY_LAST, "r") as f:
                        try: last_today = float(f.read())
                        except: last_today = round(float(res_local.get("todayWaterUsage", 0)) * 1000, 2)
                else:
                    last_today = round(float(res_local.get("todayWaterUsage", 0)) * 1000, 2)

                today_app = round(float(res_local.get("todayWaterUsage", 0)) * 1000, 2)
                delta_app = today_app - last_today if today_app >= last_today else today_app
                
                if delta_app > 0:
                    current_total += delta_app
                    current_daily_total += delta_app
                    
                    with open(CACHE_TOTAL, "w") as f: f.write(f"{current_total:.2f}")
                    with open(CACHE_DAILY_TOTAL, "w") as f: f.write(f"{current_daily_total:.2f}")
                    
                    val_m3 = round(current_total / 1000, 3)
                    influx_payload = f"woda_pomiar,okres={current_month_str} value={val_m3}"
                    try:
                        requests.post(INFLUX_URL, data=influx_payload, timeout=2)
                    except:
                        pass
                
                with open(CACHE_TODAY_LAST, "w") as f: f.write(f"{today_app:.2f}")
                
                update_domoticz(session, IDX_WODA_MIESIAC, 0, current_total)
                update_domoticz(session, IDX_WODA_DZIEN_NOWY, 0, round(current_daily_total, 1))
                
                current_water_left = res_local.get("waterLeft", 0.0)
                salt = res_local.get("saltPercent", 0)
                water_left_liters = round(float(current_water_left) * 1000, 1)
                
                update_domoticz(session, IDX_WODA_DZIS, 0, round(today_app, 1))
                update_domoticz(session, IDX_SOL, 0, salt)
                update_domoticz(session, IDX_WODA_ZAPAS, 0, water_left_liters)
            except:
                pass

        cycle_count += 1
        elapsed = time.time() - start_time
        sleep_time = max(INTERVAL_FAST - elapsed, 0.1)
        time.sleep(sleep_time)

if __name__ == "__main__":
    main()

The script is adapted to the Influx database. It creates text files that record daily and monthly water usage. As mentioned, I have a flood sensor connected, mounted next to the softening station. Additionally, you can set the flow duration in the script. After this time, the flood alarm tile also activates. This can be used for notifications in case the leak is somewhere other than the flood sensor itself. We have two polling intervals: fast (5 seconds) – water consumption, remaining liters for regeneration, regeneration date, salt level. Slow (360 cycles of 5 seconds each) – current operating mode, flood sensor (built-in), and vacation mode. Mode changes are still effective when pressed, so that's not a problem.

[Mozekam]

For now, the slow loop (server polling) is set to 30 minutes and works correctly -
you can reduce the waiting time for queries from the cloud by entering e.g.: SLOW_MULTIPLIER = 60 - then the query will be every 5 minutes.

Best regards and if you have any questions, I'm happy to help!