Skip to content

Python Decoder

Live CT-485 traffic published by the Radish device can be decoded on a host PC with mqtt_listener.py and the radish Python package. The listener subscribes to the raw MQTT topic, parses frames, and pretty-prints known message types.

You need a working ESPHome node publishing hex frames (see Software Setup) and, for useful traffic, a bus connection (see Wire and Verify).

What it does

  1. Connects to your MQTT broker using settings from .env.
  2. Subscribes to the configured topic (default radish/rs485/raw).
  3. Parses each payload into a Frame (addresses, send method, message type, payload, checksum).
  4. Dispatches the payload through MessageRegistry so known CT-485 / CT-CIM types get structured decoding.
  5. Optionally filters by message type, hides dataflow packets, and collapses adjacent routed duplicates.

Parser coverage by message ID is tracked in the Message Coverage Matrix.

Install

From the repository root:

python3 -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r requirements.txt

Current dependencies: paho-mqtt and python-dotenv

Configure .env

Copy the example and edit broker / filter settings:

cp .env.example .env
Variable Purpose
MQTT_HOST Broker hostname (default localhost)
MQTT_PORT Broker port (default 1883)
MQTT_USERNAME / MQTT_PASSWORD Auth if required; leave empty for anonymous
MQTT_TOPIC Subscribe topic (default radish/rs485/raw)
MQTT_MESSAGE_TYPES Optional comma-separated allowlist, hex or decimal (e.g. 0x87,0xC1)
MQTT_EXCLUDE_DATAFLOW true to skip dataflow packets (packet number bit 7 set), such as ACKs
MQTT_DEDUPE_REPEATED true to suppress immediately adjacent repeats of the same logical message (common when frames are relayed via the coordinator)
HA_TOKEN / HA_URL Used by request_ct_query.py to call the ESPHome get_app_query service
HA_APP_QUERY_SERVICE Optional service path override (default esphome/radish_get_app_query)
APP_QUERY_MQTT_TOPIC Topic where matching app-query response hex is published (default radish/app_query)
APP_QUERY_TIMEOUT Seconds to wait for that response (default 60)

Boolean env values accept true/false, yes/no, on/off, or 1/0. The HA_* / APP_QUERY_* variables are only needed for active queries after AutoNet join; see Requesting application queries.

Run the listener

With the venv active and .env filled in:

python mqtt_listener.py

On connect you should see subscription confirmation, then timestamped frame dumps as traffic arrives. Stop with Ctrl+C.

Example output

With polarity correct and the HVAC talking, a typical node-discovery / address-confirmation exchange looks like:

[15:43:44] radish/rs485/raw:
To: 0 (Broadcast)
From: 255 (Network Coordinator Address)
Subnet: 0 (All subnets)
Send Method: 0 (Non-Routed)
Send Parameters: 0x0000 (Requesting device's index among its node type: 0)
Source Node Type: 165 (Network Coordinator)
Message Type: 0x79 (Node Discovery Request)
Packet Number: 32 (Request/response packet; Node discovery request)
Packet Length: 1
Payload:
  Node ID Filter: 0
Checksum: 0xd93b

[15:43:47] radish/rs485/raw:
To: 0 (Broadcast)
From: 255 (Network Coordinator Address)
Subnet: 3 (>v1.0 Subordinates)
Send Method: 0 (Non-Routed)
Send Parameters: 0x0000 (Requesting device's index among its node type: 0)
Source Node Type: 165 (Network Coordinator)
Message Type: 0x76 (Address Confirmation Push Request)
Packet Number: 0 (Request/response packet; CT2.0 device)
Packet Length: 63
Payload:
  Coordinator's virtual internal Subordinate Node Type: 3 (Air Handler)
  Node type at index 1: 1 (Thermostat)
  Node type at index 2: 5 (Heat Pump)
Checksum: 0x638a

[15:43:48] radish/rs485/raw:
To: 255 (Network Coordinator Address)
From: 1 (Priority Subordinate Address, Subnet 2 or 3, thermostat or zone controller)
Subnet: 3 (>v1.0 Subordinates)
Send Method: 0 (Non-Routed)
Send Parameters: 0x0000 (Non-Routed Send Parameters from Subordinate)
Source Node Type: 1 (Thermostat)
Message Type: 0xf6 (Address Confirmation Push Response)
Packet Number: 0 (Request/response packet; CT2.0 device)
Packet Length: 63
Payload:
  Coordinator's virtual internal Subordinate Node Type: 3 (Air Handler)
  Node type at index 1: 1 (Thermostat)
  Node type at index 2: 5 (Heat Pump)
Checksum: 0xb55b

Example filter for sensor and user-menu responses only:

MQTT_MESSAGE_TYPES=0x87,0xC1
MQTT_EXCLUDE_DATAFLOW=true
MQTT_DEDUPE_REPEATED=true

If nothing prints, confirm the ESP is online, the topic matches what the device publishes, and (for bus sniffing) that raw hex is flowing — see Wire and Verify.

Package layout

Path Role
mqtt_listener.py MQTT client entrypoint
request_ct_query.py HA-triggered app query helper (config / status / sensor / id)
radish/frame.py CT-485 frame parse/encode and checksum
radish/messages/ Registered message classes (ct_485, ct_cim, MDI helpers)
radish/messages/registry.py Maps message type IDs → decoder classes
radish/maps.py Address / node-type / send-method lookups
tests/ Unit tests for frames, messages, and related contracts

Importing radish.messages registers known types as a side effect. Unknown message IDs still decode as a generic Message with a raw payload.

Request a configuration / status / sensor / identification poll

With AutoNet joined (see ESPHome Component), request_ct_query.py calls Home Assistant’s esphome.radish_get_app_query action, waits for the matching hex on APP_QUERY_MQTT_TOPIC, parses it with the same Frame decoder, prints it, and exits:

python request_ct_query.py 5 --kind config
python request_ct_query.py 3 --kind status
python request_ct_query.py 5 --kind sensor
python request_ct_query.py 5 --kind id

Requires HA_TOKEN in .env. Prefer short join windows; this is a second querier on a live bus.

Tests

python -m unittest discover -s tests -v