Michal
/
CEZ_rele_box
mirror of https://github.com/nero150/CEZ_rele_box.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add files via upload

main 0.7.0
nero150 2026-03-18 10:47:01 +01:00 committed by GitHub
parent 0826f7b897
commit bd1af21ce8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 313 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

241
custom_components/xt211_han/config_flow.py
View File

@ -1,16 +1,27 @@
"""Config flow for XT211 HAN integration."""
"""Config flow for XT211 HAN integration.
Discovery order:
1. DHCP discovery – automatic, triggered by HA when USR-DR134 appears on network
2. Network scan – user clicks "Search network" in the UI
3. Manual entry – user types IP + port manually (always available as fallback)
"""
from __future__ import annotations
import asyncio
import logging
import socket
import struct
from ipaddress import IPv4Network, IPv4Address
from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import dhcp
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_NAME
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from .const import (
DOMAIN,
@ -32,6 +43,9 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
# Known MAC prefixes for USR IOT devices (USR-DR134)
USR_IOT_MAC_PREFIXES = ("d8b04c", "b4e62d")
STEP_CONNECTION_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
@ -64,12 +78,16 @@ STEP_METER_SCHEMA = vol.Schema(
)
async def _test_connection(host: str, port: int) -> str | None:
"""Try to open a TCP connection. Returns error string or None on success."""
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
async def _test_connection(host: str, port: int, timeout: float = 5.0) -> str | None:
"""Try TCP connection. Returns error key or None on success."""
try:
reader, writer = await asyncio.wait_for(
asyncio.open_connection(host, port),
timeout=5,
timeout=timeout,
)
writer.close()
await writer.wait_closed()
@ -82,26 +100,226 @@ async def _test_connection(host: str, port: int) -> str | None:
return "unknown"
async def _scan_network(port: int, timeout: float = 0.5) -> list[str]:
"""
Scan the local network for open TCP port (default 8899).
Returns list of IP addresses that responded.
"""
# Determine local subnet from hostname
try:
local_ip = socket.gethostbyname(socket.gethostname())
except OSError:
local_ip = "192.168.1.1"
try:
network = IPv4Network(f"{local_ip}/24", strict=False)
except ValueError:
network = IPv4Network("192.168.1.0/24", strict=False)
found: list[str] = []
async def _probe(ip: str) -> None:
try:
_, writer = await asyncio.wait_for(
asyncio.open_connection(ip, port),
timeout=timeout,
)
writer.close()
await writer.wait_closed()
found.append(ip)
except Exception:
pass
# Probe all hosts in /24 concurrently (skip network and broadcast)
hosts = [str(h) for h in network.hosts()]
await asyncio.gather(*[_probe(ip) for ip in hosts])
return sorted(found)
# ---------------------------------------------------------------------------
# Config Flow
# ---------------------------------------------------------------------------
class XT211HANConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle the config flow for XT211 HAN – two steps."""
"""
Three-path config flow:
- DHCP discovery (automatic)
- Network scan (semi-automatic)
- Manual entry (always available)
"""
VERSION = 1
def __init__(self) -> None:
self._connection_data: dict[str, Any] = {}
self._discovered_host: str | None = None
self._discovered_port: int = DEFAULT_PORT
self._scan_results: list[str] = []
# ------------------------------------------------------------------
# Step 1 – connection (host / port / name)
# Path 1 – DHCP discovery (triggered automatically by HA)
# ------------------------------------------------------------------
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
"""Handle DHCP discovery of a USR IOT device."""
mac = discovery_info.macaddress.replace(":", "").lower()
if not any(mac.startswith(prefix) for prefix in USR_IOT_MAC_PREFIXES):
return self.async_abort(reason="not_supported")
ip = discovery_info.ip
_LOGGER.info("XT211 HAN: DHCP discovered USR IOT device at %s (MAC %s)", ip, mac)
# Check not already configured
await self.async_set_unique_id(f"{ip}:{DEFAULT_PORT}")
self._abort_if_unique_id_configured(updates={CONF_HOST: ip})
self._discovered_host = ip
self._discovered_port = DEFAULT_PORT
return await self.async_step_dhcp_confirm()
async def async_step_dhcp_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Ask user to confirm the DHCP-discovered device."""
if user_input is not None:
error = await _test_connection(self._discovered_host, self._discovered_port)
if error:
return self.async_show_form(
step_id="dhcp_confirm",
errors={"base": error},
description_placeholders={
"host": self._discovered_host,
"port": str(self._discovered_port),
},
)
self._connection_data = {
CONF_HOST: self._discovered_host,
CONF_PORT: self._discovered_port,
CONF_NAME: DEFAULT_NAME,
}
return await self.async_step_meter()
return self.async_show_form(
step_id="dhcp_confirm",
description_placeholders={
"host": self._discovered_host,
"port": str(self._discovered_port),
},
)
# ------------------------------------------------------------------
# Path 2 + 3 – User-initiated: scan or manual
# ------------------------------------------------------------------
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""First screen: choose between scan or manual entry."""
if user_input is not None:
if user_input.get("method") == "scan":
return await self.async_step_scan()
else:
return await self.async_step_manual()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required("method", default="scan"): vol.In(
{
"scan": "🔍 Automaticky vyhledat v síti",
"manual": "✏️ Zadat IP adresu ručně",
}
)
}
),
)
# ------------------------------------------------------------------
# Path 2 – Network scan
# ------------------------------------------------------------------
async def async_step_scan(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Scan the local network for devices with the configured port open."""
if user_input is not None:
host = user_input[CONF_HOST]
port = user_input.get(CONF_PORT, DEFAULT_PORT)
name = user_input.get(CONF_NAME, DEFAULT_NAME)
await self.async_set_unique_id(f"{host}:{port}")
self._abort_if_unique_id_configured()
error = await _test_connection(host, port)
if error:
return self.async_show_form(
step_id="scan",
data_schema=self._scan_schema(port),
errors={"base": error},
)
self._connection_data = {
CONF_HOST: host,
CONF_PORT: port,
CONF_NAME: name,
}
return await self.async_step_meter()
# Run the scan
_LOGGER.debug("XT211 HAN: scanning network for port %d", DEFAULT_PORT)
self._scan_results = await _scan_network(DEFAULT_PORT)
_LOGGER.debug("XT211 HAN: scan found %d device(s): %s", len(self._scan_results), self._scan_results)
if not self._scan_results:
# Nothing found – fall through to manual with a warning
return self.async_show_form(
step_id="scan",
data_schema=self._scan_schema(DEFAULT_PORT),
errors={"base": "no_devices_found"},
)
# Build selector: found IPs + option to type manually
choices = {ip: f"{ip}:{DEFAULT_PORT}" for ip in self._scan_results}
choices["manual"] = "✏️ Zadat jinak ručně"
return self.async_show_form(
step_id="scan",
data_schema=self._scan_schema(DEFAULT_PORT, choices),
)
def _scan_schema(
self, port: int, choices: dict | None = None
) -> vol.Schema:
if choices:
return vol.Schema(
{
vol.Required(CONF_HOST): vol.In(choices),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): str,
}
)
return vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT, default=port): int,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): str,
}
)
# ------------------------------------------------------------------
# Path 3 – Manual entry
# ------------------------------------------------------------------
async def async_step_manual(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manual IP + port entry."""
errors: dict[str, str] = {}
if user_input is not None:
host = user_input[CONF_HOST]
port = user_input[CONF_PORT]
name = user_input.get(CONF_NAME, DEFAULT_NAME)
await self.async_set_unique_id(f"{host}:{port}")
self._abort_if_unique_id_configured()
@ -110,22 +328,27 @@ class XT211HANConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if error:
errors["base"] = error
else:
self._connection_data = user_input
self._connection_data = {
CONF_HOST: host,
CONF_PORT: port,
CONF_NAME: name,
}
return await self.async_step_meter()
return self.async_show_form(
step_id="user",
step_id="manual",
data_schema=STEP_CONNECTION_SCHEMA,
errors=errors,
)
# ------------------------------------------------------------------
# Step 2 – meter configuration
# Step: meter configuration (shared by all paths)
# ------------------------------------------------------------------
async def async_step_meter(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Meter type, FVE, tariffs, relays."""
if user_input is not None:
data = {**self._connection_data, **user_input}
name = data.get(CONF_NAME, DEFAULT_NAME)

8
custom_components/xt211_han/manifest.json
View File

@ -1,12 +1,16 @@
{
"domain": "xt211_han",
"name": "XT211 HAN (RS485 via Ethernet)",
"version": "0.6.0",
"version": "0.7.0",
"documentation": "https://github.com/nero150/xt211-han-ha",
"issue_tracker": "https://github.com/nero150/xt211-han-ha/issues",
"dependencies": [],
"codeowners": ["@nero150"],
"requirements": [],
"iot_class": "local_push",
"config_flow": true
"config_flow": true,
"dhcp": [
{"macaddress": "D8B04C*"},
{"macaddress": "B4E62D*"}
]
}

28
custom_components/xt211_han/strings.json
View File

@ -2,14 +2,34 @@
"config": {
"step": {
"user": {
"title": "Nastavení připojení",
"description": "Zadej IP adresu a port RS485-to-Ethernet adaptéru (např. PUSR USR-DR134). Výchozí port pro TCP server mód je 8899.",
"title": "Přidat XT211 HAN",
"description": "Jak chceš najít adaptér?",
"data": {
"method": "Metoda vyhledání"
}
},
"scan": {
"title": "Vyhledávání v síti",
"description": "Prohledávám síť na portu 8899... Vyber nalezené zařízení nebo zadej IP ručně.",
"data": {
"host": "IP adresa adaptéru",
"port": "TCP port",
"name": "Název zařízení"
}
},
"manual": {
"title": "Ruční zadání",
"description": "Zadej IP adresu a port RS485-to-Ethernet adaptéru (např. PUSR USR-DR134). Výchozí port je 8899.",
"data": {
"host": "IP adresa adaptéru",
"port": "TCP port",
"name": "Název zařízení"
}
},
"dhcp_confirm": {
"title": "Nalezeno zařízení USR IOT",
"description": "Home Assistant automaticky nalezl adaptér na adrese **{host}:{port}**. Chceš nastavit integraci XT211 HAN pro toto zařízení?"
},
"meter": {
"title": "Konfigurace elektroměru",
"description": "Upřesni parametry tvého elektroměru. Podle toho se zobrazí jen relevantní entity.",
@ -23,10 +43,12 @@
},
"error": {
"cannot_connect": "Nepodařilo se připojit k adaptéru. Zkontroluj IP adresu, port a připojení.",
"no_devices_found": "V síti nebyla nalezena žádná zařízení na portu 8899. Zkus zadat IP adresu ručně.",
"unknown": "Neočekávaná chyba. Zkontroluj log."
},
"abort": {
"already_configured": "Toto zařízení je již nakonfigurováno."
"already_configured": "Toto zařízení je již nakonfigurováno.",
"not_supported": "Toto zařízení není podporováno."
}
}
}

28
custom_components/xt211_han/translations/cs.json
View File

@ -2,14 +2,34 @@
"config": {
"step": {
"user": {
"title": "Nastavení připojení",
"description": "Zadej IP adresu a port RS485-to-Ethernet adaptéru (např. PUSR USR-DR134). Výchozí port pro TCP server mód je 8899.",
"title": "Přidat XT211 HAN",
"description": "Jak chceš najít adaptér?",
"data": {
"method": "Metoda vyhledání"
}
},
"scan": {
"title": "Vyhledávání v síti",
"description": "Prohledávám síť na portu 8899... Vyber nalezené zařízení nebo zadej IP ručně.",
"data": {
"host": "IP adresa adaptéru",
"port": "TCP port",
"name": "Název zařízení"
}
},
"manual": {
"title": "Ruční zadání",
"description": "Zadej IP adresu a port RS485-to-Ethernet adaptéru (např. PUSR USR-DR134). Výchozí port je 8899.",
"data": {
"host": "IP adresa adaptéru",
"port": "TCP port",
"name": "Název zařízení"
}
},
"dhcp_confirm": {
"title": "Nalezeno zařízení USR IOT",
"description": "Home Assistant automaticky nalezl adaptér na adrese **{host}:{port}**. Chceš nastavit integraci XT211 HAN pro toto zařízení?"
},
"meter": {
"title": "Konfigurace elektroměru",
"description": "Upřesni parametry tvého elektroměru. Podle toho se zobrazí jen relevantní entity.",
@ -23,10 +43,12 @@
},
"error": {
"cannot_connect": "Nepodařilo se připojit k adaptéru. Zkontroluj IP adresu, port a připojení.",
"no_devices_found": "V síti nebyla nalezena žádná zařízení na portu 8899. Zkus zadat IP adresu ručně.",
"unknown": "Neočekávaná chyba. Zkontroluj log."
},
"abort": {
"already_configured": "Toto zařízení je již nakonfigurováno."
"already_configured": "Toto zařízení je již nakonfigurováno.",
"not_supported": "Toto zařízení není podporováno."
}
}
}

28
custom_components/xt211_han/translations/en.json
View File

@ -2,14 +2,34 @@
"config": {
"step": {
"user": {
"title": "Connection Setup",
"description": "Enter the IP address and port of your RS485-to-Ethernet adapter (e.g. PUSR USR-DR134). The default TCP server port is 8899.",
"title": "Add XT211 HAN",
"description": "How would you like to find the adapter?",
"data": {
"method": "Discovery method"
}
},
"scan": {
"title": "Network Scan",
"description": "Scanning network on port 8899... Select a discovered device or enter IP manually.",
"data": {
"host": "Adapter IP address",
"port": "TCP port",
"name": "Device name"
}
},
"manual": {
"title": "Manual Entry",
"description": "Enter the IP address and port of your RS485-to-Ethernet adapter (e.g. PUSR USR-DR134). Default port is 8899.",
"data": {
"host": "Adapter IP address",
"port": "TCP port",
"name": "Device name"
}
},
"dhcp_confirm": {
"title": "USR IOT Device Found",
"description": "Home Assistant automatically discovered an adapter at **{host}:{port}**. Would you like to set up the XT211 HAN integration for this device?"
},
"meter": {
"title": "Meter Configuration",
"description": "Specify your meter parameters. Only relevant entities will be shown.",
@ -23,10 +43,12 @@
},
"error": {
"cannot_connect": "Could not connect to the adapter. Check the IP address, port and network connection.",
"no_devices_found": "No devices found on port 8899. Try entering the IP address manually.",
"unknown": "Unexpected error. Check the log."
},
"abort": {
"already_configured": "This device is already configured."
"already_configured": "This device is already configured.",
"not_supported": "This device is not supported."
}
}
}