I wanted to post the captured data from my spa here, but there's so much of it that processing errors occurred, so I decided against it to avoid unnecessary confusion. Instead, I started analyzing the material.
christopheknap
My approach to the ozonator and the 0xC1 flag was incorrect. Of course, it worked for me, but not necessarily for others.
And not only that, but you were right that it's definitely better to work with individual bits instead of whole byte values. I've modified my script, starting byte numbering from 0.
ozone - checking bit 0 and/or bit 7 in byte 14 will be correct.
LED - bit 0 in byte 17.
filtering - bit 1 in byte 12.
massage - bit 2 in byte 12.
heating: circulation bit 2 = 0, bit 4 = 1 in byte 14.
heating bit 2 = 1, bit 4 = 1 in byte 14.
cooldown - I have a problem here. You can read it from my code. It works for me, but I don't know if it will be portable to other configurations.
I've fixed another small bug. If anyone is interested, feel free to test it.
old-man
You can connect
My new script:
import socket
import json
import time
import paho.mqtt.client as mqtt
# Configuration constants
SPA_IP = "ADD_IP_SPA"
SPA_PORT = 8899
MQTT_HOST = "ADD_IP_MQQT (HA)"
MQTT_PORT = 1883
# Initialize MQTT Client using the updated API version
mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.username_pw_set("user_mqtt", "passw_mtt")
def f_to_c(f):
"""Convert Fahrenheit to Celsius and round to 1 decimal place."""
return round((f - 32) * 5 / 9, 1)
def pseudo_unescape(data: bytes) -> bytes:
"""
Handles byte-unstuffing for RS485 communication.
Replaces 2-byte escape sequences starting with 0x1B back into original raw bytes.
"""
out = bytearray()
i = 0
while i < len(data):
if i + 1 < len(data) and data[i] == 0x1B:
nxt = data[i + 1]
if nxt == 0x11:
out.append(0x1A)
i += 2
continue
elif nxt == 0x13:
out.append(0x1C)
i += 2
continue
elif nxt == 0x14:
out.append(0x1D)
i += 2
continue
elif nxt == 0x15:
out.append(0x1E)
i += 2
continue
elif nxt == 0x0B:
out.append(0x1B)
i += 2
continue
out.append(data[i])
i += 1
return bytes(out)
def parse_frame(frame):
"""Parses telemetry data from the unescaped RS485 frame and publishes it via MQTT."""
data = {}
try:
# Extract temperatures
temp_f = frame[9]
set_f = frame[16]
data["water_temp"] = f_to_c(temp_f)
data["target_temp"] = f_to_c(set_f)
# Light - Check bit 0 in byte 17 (Mask: 0x01 -> 00000001)
data["light"] = bool(frame[17] & 0x01)
# Filtering - Check bit 1 in both byte 12 and byte 27 (Mask: 0x02 -> 00000010)
data["filtering"] = (
bool(frame[12] & 0x02) and bool(frame[27] & 0x02)
)
# Massage - Check bit 2 in both byte 12 and byte 27 (Mask: 0x04 -> 00000100)
data["massage"] = (
bool(frame[12] & 0x04) and bool(frame[27] & 0x04)
)
# Heating state logic based on specific bits
f14 = frame[14]
f17 = frame[17]
# Circulation: in byte 14, bit 4 == 1 AND bit 2 == 0
if (f14 & 0x10) != 0 and (f14 & 0x04) == 0:
data["heating_state"] = "circulation"
# Heating: in byte 14, bit 4 == 1 AND bit 2 == 1 (Combined mask: 0x14 -> 00010100)
elif (f14 & 0x14) == 0x14:
data["heating_state"] = "heating"
# Cooldown: in byte 14, bit 0 == 0 AND bit 4 == 0, and in byte 17, bit 7 == 1 (Mask: 0x80)
elif (
(f14 & 0x01) == 0 and
(f14 & 0x10) == 0 and
bool(f17 & 0x80)
):
data["heating_state"] = "cooldown"
else:
data["heating_state"] = "off"
# UV - Check if both bit 0 and bit 7 in byte 14 are set to 1 (Combined mask: 0x81 -> 10000001)
data["uv"] = (
(f14 & 0x81) == 0x81 and
frame[28] == 0x20
)
# Datetime extraction
year = 2000 + frame[53]
month = frame[54]
day = frame[55]
hour = frame[56]
minute = frame[57]
second = frame[58]
data["spa_time"] = (
f"{year:04d}-{month:02d}-{day:02d} "
f"{hour:02d}:{minute:02d}:{second:02d}"
)
# Weekday mapping from byte 59 (0 = Sunday, 1 = Monday, etc.)
weekdays = {
0: "Sunday",
1: "Monday",
2: "Tuesday",
3: "Wednesday",
4: "Thursday",
5: "Friday",
6: "Saturday"
}
raw_weekday = frame[59]
data["spa_weekday"] = weekdays.get(raw_weekday, "unknown")
# Publish the finalized data dictionary as a JSON payload to Home Assistant / MQTT
info = mqttc.publish(
"spa/state",
json.dumps(data),
retain=True
)
print("MQTT RC:", info.rc)
print(data)
except Exception as e:
print("Parse error:", e)
def main():
# Connect to the MQTT Broker
mqttc.connect(MQTT_HOST, MQTT_PORT, 60)
# FIX: Start the background thread for MQTT only ONCE before entering the loop.
# This prevents creating multiple infinite background threads on socket drops.
mqttc.loop_start()
print("MQTT background thread started successfully")
while True:
try:
# Create a TCP socket to communicate with the RS485-to-Ethernet bridge
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((SPA_IP, SPA_PORT))
print("Connected to spa")
buffer = bytearray()
while True:
# Receive raw chunk of data from the socket stream
chunk = sock.recv(1024)
if not chunk:
raise Exception("Disconnected")
buffer.extend(chunk)
# Process all complete frames currently available in the buffer
while True:
try:
# Locate frame boundaries using markers (0x1A = Start, 0x1D = End)
start = buffer.index(0x1A)
end = buffer.index(0x1D, start)
except ValueError:
# Break inner loop if a full frame boundary is not yet available
break
# Extract the standalone raw frame
frame = bytes(buffer[start:end + 1])
# Clear the processed frame (including any preceding noise) from the main buffer
del buffer[:end + 1]
# Sanity check on minimum frame length
if len(frame) < 10:
continue
# Verify frame routing identifier
if frame[1] != 0xFF:
continue
# Remove byte-stuffing overhead before parsing fields
frame = pseudo_unescape(frame)
# Extract payload and ship via MQTT
parse_frame(frame)
except Exception as e:
print("Connection error:", e)
# Wait 5 seconds before attempting to reconnect to the spa
time.sleep(5)
if __name__ == "__main__":
main()
And mqtt.yaml
sensor:
- name: "Spa Water Temperature"
state_topic: "spa/state"
value_template: "{{ value_json.water_temp }}"
unit_of_measurement: "°C"
device_class: "temperature"
state_class: "measurement"
unique_id: "spa_water_temperature"
- name: "Spa Target Temperature"
state_topic: "spa/state"
value_template: "{{ value_json.target_temp }}"
unit_of_measurement: "°C"
device_class: "temperature"
unique_id: "spa_target_temperature"
- name: "Spa Heating State"
state_topic: "spa/state"
value_template: "{{ value_json.heating_state }}"
icon: "mdi:water-boiler"
unique_id: "spa_heating_state"
- name: "Spa Controller Time"
state_topic: "spa/state"
value_template: "{{ value_json.spa_time }}"
icon: "mdi:clock-digital"
unique_id: "spa_controller_time"
- name: "Spa Day of Week"
state_topic: "spa/state"
value_template: "{{ value_json.spa_weekday }}"
icon: "mdi:calendar-range"
unique_id: "spa_day_of_week"
binary_sensor:
- name: "Spa Light"
state_topic: "spa/state"
value_template: "{{ 'ON' if value_json.light else 'OFF' }}"
device_class: "light"
unique_id: "spa_light"
- name: "Spa Filtering"
state_topic: "spa/state"
value_template: "{{ 'ON' if value_json.filtering else 'OFF' }}"
device_class: "running"
icon: "mdi:filter"
unique_id: "spa_filtering"
- name: "Spa Massage"
state_topic: "spa/state"
value_template: "{{ 'ON' if value_json.massage else 'OFF' }}"
device_class: "running"
icon: "mdi:hydro-power"
unique_id: "spa_massage"
- name: "Spa UV Sterilizer"
state_topic: "spa/state"
value_template: "{{ 'ON' if value_json.uv else 'OFF' }}"
device_class: "safety"
icon: "mdi:lightbulb-uv"
unique_id: "spa_uv_sterilizer"
Alpha state: The integration is not yet officially released. Some write commands still need live verification at the spa. That said, testers are very welcome! If you have a P25B85 + PB554 and an EW11 or W610, feel free to install via HACS custom repo and report back.
I sincerely apologize that the integration corrupted your spa configuration. That should absolutely never happen. I'm treating this as the highest priority issue.