public_git
/
ElVit-esphome_components
mirror of https://github.com/ElVit/esphome_components.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #18 from ElVit/heatpump_uart_client_fix 

Heatpump uart client timeout fix (fixes #13)
This commit is contained in:
ElVit 2025-12-23 11:19:03 +01:00 committed by GitHub
parent c3c6106ade e28ffa3288
commit 75f1f267ba
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 233 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
components/panasonic_heatpump/README.md
View File

@ -62,6 +62,7 @@ panasonic_heatpump:
uart_client_id: uart_cz_taw1
log_uart_msg: true
update_interval: 5s
uart_client_timeout: 10s
sensor:
- platform: panasonic_heatpump
@ -109,15 +110,16 @@ climate:
## Configuration variables
- **id** (*Optional*, ID): Manually specify the ID used for actions.
- **update_interval** (*Optional*, Time): The interval of the polling request message to get the heatpump values. This time applies only if no `uart_client_id` is set. Defaults to 3s.
- **uart_id** (Optional, ID): Manually specify the UART ID of the Heatpump. Required if multiple UART buses are defined.
- **uart_client_id** (*Optional*, ID): Manually specify the UART ID of an additonal UART client like the Panasonic CZ-TAW1. If this ID is not set then your ESP controller will send the polling request messages.
- **log_uart_msg** (*Optional*, boolean): Shows the raw UART messages in the logs, if set to `true`. The messages will be written to the log level `INFO`. Defaults to false.
* **id** (*Optional*, ID): Manually specify the ID used for actions.
* **update_interval** (*Optional*, [Time](https://esphome.io/guides/configuration-types/#time)): The interval of the polling request message to get the heatpump values. This time applies only if no `uart_client_id` is set. Defaults to `3s`.
* **uart_id** (Optional, [ID](https://esphome.io/guides/configuration-types/#id)): Manually specify the UART ID of the Heatpump. Required if multiple UART buses are defined.
* **uart_client_id** (*Optional*, [ID](https://esphome.io/guides/configuration-types/#id)): Manually specify the UART ID of an additonal UART client like the Panasonic CZ-TAW1. If this ID is not set then your ESP controller will send the polling request messages.
* **log_uart_msg** (*Optional*, boolean): Shows the raw UART messages in the logs, if set to `true`. The messages will be written to the log level `INFO`. Defaults to false.
* **uart_client_timeout** (*Optional*, [Time](https://esphome.io/guides/configuration-types/#time)): Maximum idle time for the `uart_client_id` connection. If no request is sent from the UART client for this duration, the component will send a POLLING request to the heatpump. This will (hopefully) prevent the heatpump from becoming unresponsive. This feature can also be disabled by setting this option to `100ms` or lower. Defaults to `10s`.
### Sensors
All sensors are optional and all default sensor variables can be applied.
All sensors are optional and all options from [sensor component](https://esphome.io/components/sensor/) can be applied.
Here a list of all supported sensors:
```yaml
@ -330,12 +332,14 @@ sensor:
### Binary Sensors
All binary sensors are optional and all default binary sensor variables can be applied.
All binary sensors are optional and all options from [binary sensor component](https://esphome.io/components/binary_sensor/) can be applied.
Here a list of all supported binary sensors:
```yaml
binary_sensor:
- platform: panasonic_heatpump
uart_client_timed_out:
name: "UART Client Timed Out"
top0:
name: "Heatpump State"
top2:
@ -386,7 +390,7 @@ binary_sensor:
### Text Sensors
All text sensors are optional and all default text sensor variables can be applied.
All text sensors are optional and all options from [text sensor component](https://esphome.io/components/text_sensor/) can be applied.
Here a list of all supported text sensors:
```yaml
@ -438,7 +442,7 @@ text_sensor:
### Numbers
All numbers are optional and all default number variables can be applied.
All numbers are optional and all options from [number component](https://esphome.io/components/number/) can be applied.
Additionally the options `min_value`, `max_value` and `step` can override the default limits of each set entitiy.
This is usefull for example for `set5` to `set8` if `direct temperature` is configured instead of `compensation curve` (see `top76` and `top81`).
Here a list of all supported numbers:
@ -516,7 +520,7 @@ number:
### Switches
All switches are optional and all default switch variables can be applied.
All switches are optional and all options from [switch component](https://esphome.io/components/switch/) can be applied.
Here a list of all supported switches:
```yaml
@ -552,7 +556,7 @@ switch:
### Selects
All selects are optional and all default select variables can be applied.
All selects are optional and all options from [select component](https://esphome.io/components/select/) can be applied.
Additionally the option `cool_mode` can be configured.
If `cool_mode` is set to `true` the entity `set9` will have the additional select options `COOL`, `COOL+TANK`, `AUTO` and `AUTO+TANK`.
Here a list of all supported selects:
@ -579,7 +583,7 @@ select:
### Climates
All climates are optional and all default climate variables can be applied.
All climates are optional and all options from [climate component](https://esphome.io/components/climate/) can be applied.
Additionally the option `cool_mode` can be configured.
If `cool_mode` is set to `true` the entity `zone1` and `zone2` will have the additional climate modes `COOL` and `AUTO`.
Additionally the options `min_temperature`, `max_temperature` and `temperature_step` can override the default limits on each climate entitiy.

5
components/panasonic_heatpump/__init__.py
View File

@ -9,6 +9,7 @@ MULTICONF = True
CONF_PANASONIC_HEATPUMP_ID = "panasonic_heatpump"
CONF_UART_CLIENT = "uart_client_id"
CONF_UART_CLIENT_TIMEOUT = "uart_client_timeout"
CONF_LOG_UART_MSG = "log_uart_msg"
panasonic_heatpump_ns = cg.esphome_ns.namespace("panasonic_heatpump")
@ -21,6 +22,9 @@ CONFIG_SCHEMA = (
{
cv.GenerateID(): cv.declare_id(PanasonicHeatpumpComponent),
cv.Optional(CONF_UART_CLIENT): cv.use_id(uart.UARTComponent),
cv.Optional(
CONF_UART_CLIENT_TIMEOUT, default="10000ms"
): cv.positive_time_period_milliseconds,
cv.Optional(CONF_LOG_UART_MSG, default=False): cv.boolean,
}
)
@ -37,5 +41,6 @@ async def to_code(config):
if CONF_UART_CLIENT in config:
uart_client = await cg.get_variable(config[CONF_UART_CLIENT])
cg.add(var.set_uart_client(uart_client))
cg.add(var.set_uart_client_timeout(config[CONF_UART_CLIENT_TIMEOUT]))
cg.add(var.set_log_uart_msg(config[CONF_LOG_UART_MSG]))

5
components/panasonic_heatpump/binary_sensor/__init__.py
View File

@ -10,6 +10,7 @@ from .. import (
panasonic_heatpump_ns,
)
CONF_UART_CLIENT_TIMED_OUT = "uart_client_timed_out"
CONF_TOP0 = "top0" # Heatpump State
CONF_TOP2 = "top2" # Force DHW State
CONF_TOP3 = "top3" # Quiet Mode Schedule
@ -35,6 +36,7 @@ CONF_TOP132 = "top132" # Bivalent Advanced Heat
CONF_TOP133 = "top133" # Bivalent Advanced DHW
TYPES = [
CONF_UART_CLIENT_TIMED_OUT,
CONF_TOP0,
CONF_TOP2,
CONF_TOP3,
@ -69,6 +71,9 @@ CONFIG_SCHEMA = cv.Schema(
cv.GenerateID(CONF_PANASONIC_HEATPUMP_ID): cv.use_id(
PanasonicHeatpumpComponent
),
cv.Optional(CONF_UART_CLIENT_TIMED_OUT): binary_sensor.binary_sensor_schema(
PanasonicHeatpumpBinarySensor,
),
cv.Optional(CONF_TOP0): binary_sensor.binary_sensor_schema(
PanasonicHeatpumpBinarySensor,
device_class=DEVICE_CLASS_RUNNING,

5
components/panasonic_heatpump/binary_sensor/panasonic_heatpump_binary_sensor.cpp
View File

@ -15,6 +15,11 @@ void PanasonicHeatpumpBinarySensor::publish_new_state(const std::vector<uint8_t>
bool new_state;
switch (this->id_) {
case BinarySensorIds::CONF_UART_CLIENT_TIMED_OUT:
new_state = this->parent_->get_uart_client_timeout_exceeded();
if (this->has_state() && this->state == new_state)
return;
break;
case BinarySensorIds::CONF_TOP0:
new_state = PanasonicDecode::getBinaryState(PanasonicDecode::getBit7and8(data[4]));
if (this->has_state() && this->state == new_state)

1
components/panasonic_heatpump/binary_sensor/panasonic_heatpump_binary_sensor.h
View File

@ -7,6 +7,7 @@
namespace esphome {
namespace panasonic_heatpump {
enum BinarySensorIds : uint8_t {
CONF_UART_CLIENT_TIMED_OUT,
CONF_TOP0,
CONF_TOP2,
CONF_TOP3,

30
components/panasonic_heatpump/panasonic_heatpump.cpp
View File

@ -25,6 +25,15 @@ void PanasonicHeatpumpComponent::update() {
}
void PanasonicHeatpumpComponent::loop() {
// Check if no request was sent for uart_client_timeout when uart_client is configured
if (this->uart_client_ != nullptr && this->uart_client_timeout_ > 100) {
uint32_t current_time = millis();
if (current_time - this->last_request_time_ >= this->uart_client_timeout_) {
this->next_request_ = RequestType::POLLING;
this->uart_client_timeout_exceeded_ = true;
}
}
switch (this->loop_state_) {
case LoopState::READ_RESPONSE:
this->read_response();
@ -132,15 +141,15 @@ void PanasonicHeatpumpComponent::read_response() {
if (this->response_message_.size() == 3 && byte_ != 0x01 && byte_ != 0x10) {
this->response_receiving_ = false;
ESP_LOGW(TAG, "Invalid response message: 0x%s. Expected last byte to be 0x01 or 0x10",
PanasonicHelpers::byte_array_to_hex_string(this->response_message_, ','));
PanasonicHelpers::byte_array_to_hex_string(this->response_message_, ',').c_str());
delay(10); // NOLINT
continue;
}
// 4. byte shall be 0x10 or 0x21
if (this->response_message_.size() == 4 && byte_ != 0x10 && byte_ != 0x21) {
// 4. byte shall be 0x01, 0x10 or 0x21
if (this->response_message_.size() == 4 && byte_ != 0x01 && byte_ != 0x10 && byte_ != 0x21) {
this->response_receiving_ = false;
ESP_LOGW(TAG, "Invalid response message: 0x%s. Expected last byte to be 0x10 or 0x21",
PanasonicHelpers::byte_array_to_hex_string(this->response_message_, ','));
ESP_LOGW(TAG, "Invalid response message: 0x%s. Expected last byte to be 0x01, 0x10 or 0x21",
PanasonicHelpers::byte_array_to_hex_string(this->response_message_, ',').c_str());
delay(10); // NOLINT
continue;
}
@ -184,6 +193,9 @@ void PanasonicHeatpumpComponent::send_request(RequestType requestType) {
break;
};
// Update last request time when request was sent
this->last_request_time_ = millis();
this->next_request_ = RequestType::NONE;
}
@ -214,7 +226,7 @@ void PanasonicHeatpumpComponent::read_request() {
if (this->request_message_.size() == 3 && byte_ != 0x01 && byte_ != 0x10) {
this->request_receiving_ = false;
ESP_LOGW(TAG, "Invalid request message: 0x%s. Expected last byte to be 0x01 or 0x10",
PanasonicHelpers::byte_array_to_hex_string(this->request_message_, ','));
PanasonicHelpers::byte_array_to_hex_string(this->request_message_, ',').c_str());
delay(10); // NOLINT
continue;
}
@ -222,7 +234,7 @@ void PanasonicHeatpumpComponent::read_request() {
if (this->request_message_.size() == 4 && byte_ != 0x10 && byte_ != 0x21) {
this->request_receiving_ = false;
ESP_LOGW(TAG, "Invalid request message: 0x%s. Expected last byte to be 0x10 or 0x21",
PanasonicHelpers::byte_array_to_hex_string(this->request_message_, ','));
PanasonicHelpers::byte_array_to_hex_string(this->request_message_, ',').c_str());
delay(10); // NOLINT
continue;
}
@ -232,6 +244,10 @@ void PanasonicHeatpumpComponent::read_request() {
this->request_receiving_ = false;
if (this->log_uart_msg_)
PanasonicHelpers::log_uart_hex(UART_LOG_TX, this->request_message_, ',');
// Update last request time when request is complete
this->last_request_time_ = millis();
this->uart_client_timeout_exceeded_ = false;
}
}
}

11
components/panasonic_heatpump/panasonic_heatpump.h
View File

@ -12,7 +12,7 @@
#include "commands.h"
#ifndef PANASONIC_HEATPUMP_VERSION
#define PANASONIC_HEATPUMP_VERSION "0.0.5"
#define PANASONIC_HEATPUMP_VERSION "0.0.5-beta.2"
#endif
namespace esphome {
@ -74,6 +74,9 @@ class PanasonicHeatpumpComponent : public PollingComponent, public uart::UARTDev
void set_uart_client(uart::UARTComponent* uart) {
this->uart_client_ = uart;
}
void set_uart_client_timeout(uint32_t timeout_ms) {
this->uart_client_timeout_ = timeout_ms;
}
void set_log_uart_msg(bool active) {
this->log_uart_msg_ = active;
}
@ -110,11 +113,16 @@ class PanasonicHeatpumpComponent : public PollingComponent, public uart::UARTDev
void add_extra_sensor(PanasonicHeatpumpEntity* sensor) {
extra_sensors_.push_back(sensor);
}
bool get_uart_client_timeout_exceeded() {
return this->uart_client_timeout_exceeded_;
}
protected:
// options variables
uart::UARTComponent* uart_client_{nullptr};
bool log_uart_msg_{false};
uint32_t last_request_time_{0};
uint32_t uart_client_timeout_{10000};
// uart message variables
std::vector<uint8_t> heatpump_default_message_;
std::vector<uint8_t> heatpump_extra_message_;
@ -128,6 +136,7 @@ class PanasonicHeatpumpComponent : public PollingComponent, public uart::UARTDev
bool response_receiving_{false};
bool request_receiving_{false};
bool send_extra_request_{false};
bool uart_client_timeout_exceeded_{false};
LoopState loop_state_{LoopState::RESTART_LOOP};
RequestType next_request_{RequestType::INITIAL};
ResponseType current_response_{ResponseType::UNKNOWN};

2
components/panasonic_heatpump/text_sensor/__init__.py
View File

@ -24,7 +24,7 @@ ICON_PUMP = "mdi:pump"
ICON_EXTERNAL_PAD_HEATER = "mdi:radiator"
CONF_TOP4 = "top4" # Operation Mode
# ToDo: Split up top4 into top4_1 (Heating Mode State) and top4_2 (DHW Mode State)
# TODO: Split up top4 into top4_1 (Heating Mode State) and top4_2 (DHW Mode State)
CONF_TOP17 = "top17" # Powerful Mode Time
CONF_TOP18 = "top18" # Quiet Mode Level
CONF_TOP19 = "top19" # Holiday Mode State

87
tests/panasonic_heatpump/test_panasonic_heatpump_integration.py
View File

@ -16,121 +16,154 @@ class TestPanasonicHeatpumpIntegration:
def test_yaml_esp8266(self):
"""Get the path to the ESP8266 test YAML file."""
base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
return os.path.join(base_dir, 'tests', 'panasonic_heatpump', 'test_panasonic_heatpump_esp8266.yaml')
return os.path.join(
base_dir,
"tests",
"panasonic_heatpump",
"test_panasonic_heatpump_esp8266.yaml",
)
@pytest.fixture
def test_yaml_esp32(self):
"""Get the path to the ESP32 test YAML file."""
base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
return os.path.join(base_dir, 'tests', 'panasonic_heatpump', 'test_panasonic_heatpump_full.yaml')
return os.path.join(
base_dir, "tests", "panasonic_heatpump", "test_panasonic_heatpump_full.yaml"
)
@pytest.fixture
def test_yaml_esp32s2(self):
"""Get the path to the ESP32-S2 test YAML file."""
base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
return os.path.join(base_dir, 'tests', 'panasonic_heatpump', 'test_panasonic_heatpump_esp32s2.yaml')
return os.path.join(
base_dir,
"tests",
"panasonic_heatpump",
"test_panasonic_heatpump_esp32s2.yaml",
)
@pytest.fixture
def test_yaml_esp32c3(self):
"""Get the path to the ESP32-C3 test YAML file."""
base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
return os.path.join(base_dir, 'tests', 'panasonic_heatpump', 'test_panasonic_heatpump_esp32c3.yaml')
return os.path.join(
base_dir,
"tests",
"panasonic_heatpump",
"test_panasonic_heatpump_esp32c3.yaml",
)
@pytest.fixture
def test_yaml_cztaw1(self):
"""Get the path to the CZ-TAW1 client test YAML file."""
base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
return os.path.join(base_dir, 'tests', 'panasonic_heatpump', 'test_panasonic_heatpump_cztaw1.yaml')
return os.path.join(
base_dir,
"tests",
"panasonic_heatpump",
"test_panasonic_heatpump_cztaw1.yaml",
)
def test_validate_esp8266_config(self, test_yaml_esp8266):
def test_validate_esp8266_config(self, test_yaml_esp8266: str):
"""Test that the ESP8266 configuration is valid."""
if not os.path.exists(test_yaml_esp8266):
pytest.skip(f"Test file not found: {test_yaml_esp8266}")
try:
result = subprocess.run(
['esphome', 'config', test_yaml_esp8266],
["esphome", "config", test_yaml_esp8266],
capture_output=True,
text=True,
timeout=120
timeout=120,
)
assert result.returncode == 0, f"ESP8266 config validation failed: {result.stderr}"
assert (
result.returncode == 0
), f"ESP8266 config validation failed: {result.stderr}"
except FileNotFoundError:
pytest.skip("ESPHome not installed")
except subprocess.TimeoutExpired:
pytest.fail("ESPHome config validation timed out")
def test_validate_esp32_config(self, test_yaml_esp32):
def test_validate_esp32_config(self, test_yaml_esp32: str):
"""Test that the ESP32 configuration is valid."""
if not os.path.exists(test_yaml_esp32):
pytest.skip(f"Test file not found: {test_yaml_esp32}")
try:
result = subprocess.run(
['esphome', 'config', test_yaml_esp32],
["esphome", "config", test_yaml_esp32],
capture_output=True,
text=True,
timeout=120
timeout=120,
)
assert result.returncode == 0, f"ESP32 config validation failed: {result.stderr}"
assert (
result.returncode == 0
), f"ESP32 config validation failed: {result.stderr}"
except FileNotFoundError:
pytest.skip("ESPHome not installed")
except subprocess.TimeoutExpired:
pytest.fail("ESPHome config validation timed out")
def test_validate_esp32s2_config(self, test_yaml_esp32s2):
def test_validate_esp32s2_config(self, test_yaml_esp32s2: str):
"""Test that the ESP32-S2 configuration is valid."""
if not os.path.exists(test_yaml_esp32s2):
pytest.skip(f"Test file not found: {test_yaml_esp32s2}")
try:
result = subprocess.run(
['esphome', 'config', test_yaml_esp32s2],
["esphome", "config", test_yaml_esp32s2],
capture_output=True,
text=True,
timeout=120
timeout=120,
)
assert result.returncode == 0, f"ESP32-S2 config validation failed: {result.stderr}"
assert (
result.returncode == 0
), f"ESP32-S2 config validation failed: {result.stderr}"
except FileNotFoundError:
pytest.skip("ESPHome not installed")
except subprocess.TimeoutExpired:
pytest.fail("ESPHome config validation timed out")
def test_validate_esp32c3_config(self, test_yaml_esp32c3):
def test_validate_esp32c3_config(self, test_yaml_esp32c3: str):
"""Test that the ESP32-C3 configuration is valid."""
if not os.path.exists(test_yaml_esp32c3):
pytest.skip(f"Test file not found: {test_yaml_esp32c3}")
try:
result = subprocess.run(
['esphome', 'config', test_yaml_esp32c3],
["esphome", "config", test_yaml_esp32c3],
capture_output=True,
text=True,
timeout=120
timeout=120,
)
assert result.returncode == 0, f"ESP32-C3 config validation failed: {result.stderr}"
assert (
result.returncode == 0
), f"ESP32-C3 config validation failed: {result.stderr}"
except FileNotFoundError:
pytest.skip("ESPHome not installed")
except subprocess.TimeoutExpired:
pytest.fail("ESPHome config validation timed out")
def test_validate_cztaw1_config(self, test_yaml_cztaw1):
def test_validate_cztaw1_config(self, test_yaml_cztaw1: str):
"""Test that the CZ-TAW1 client configuration with UART-proxy is valid."""
if not os.path.exists(test_yaml_cztaw1):
pytest.skip(f"Test file not found: {test_yaml_cztaw1}")
try:
result = subprocess.run(
['esphome', 'config', test_yaml_cztaw1],
["esphome", "config", test_yaml_cztaw1],
capture_output=True,
text=True,
timeout=120
timeout=120,
)
assert result.returncode == 0, f"CZ-TAW1 config validation failed: {result.stderr}"
assert (
result.returncode == 0
), f"CZ-TAW1 config validation failed: {result.stderr}"
except FileNotFoundError:
pytest.skip("ESPHome not installed")
except subprocess.TimeoutExpired:
pytest.fail("ESPHome config validation timed out")
if __name__ == '__main__':
pytest.main([__file__, '-v'])
if __name__ == "__main__":
pytest.main([__file__, "-v"])

191
tests/panasonic_heatpump/test_panasonic_heatpump_unit.py
View File

@ -6,7 +6,6 @@ Tests the Python configuration schema and validation logic.
import pytest
from unittest.mock import Mock, MagicMock, patch
import esphome.config_validation as cv
from esphome.core import CORE
@ -23,13 +22,13 @@ class TestPanasonicHeatpumpConfig:
@pytest.fixture
def mock_uart(self):
"""Mock UART component."""
with patch('esphome.components.uart') as mock:
with patch("esphome.components.uart") as mock:
yield mock
@pytest.fixture
def mock_cg(self):
"""Mock code generator."""
with patch('esphome.codegen') as mock:
with patch("esphome.codegen") as mock:
mock.esphome_ns = MagicMock()
mock.esphome_ns.namespace.return_value.class_ = MagicMock()
yield mock
@ -39,13 +38,15 @@ class TestPanasonicHeatpumpConfig:
try:
import sys
import os
# Add components directory to path
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
assert ph_init is not None
except ImportError as e:
pytest.fail(f"Failed to import panasonic_heatpump: {e}")
@ -54,95 +55,99 @@ class TestPanasonicHeatpumpConfig:
"""Test component metadata is correctly defined."""
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
assert hasattr(ph_init, 'CODEOWNERS')
assert ph_init.CODEOWNERS == ['@elvit']
assert hasattr(ph_init, 'MULTICONF')
assert hasattr(ph_init, "CODEOWNERS")
assert ph_init.CODEOWNERS == ["@elvit"]
assert hasattr(ph_init, "MULTICONF")
assert ph_init.MULTICONF is True
assert hasattr(ph_init, 'DEPENDENCIES')
assert 'uart' in ph_init.DEPENDENCIES
assert hasattr(ph_init, "DEPENDENCIES")
assert "uart" in ph_init.DEPENDENCIES
def test_config_constants(self):
"""Test that configuration constants are defined."""
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
assert hasattr(ph_init, 'CONF_PANASONIC_HEATPUMP_ID')
assert hasattr(ph_init, "CONF_PANASONIC_HEATPUMP_ID")
assert ph_init.CONF_PANASONIC_HEATPUMP_ID == "panasonic_heatpump"
assert hasattr(ph_init, 'CONF_UART_CLIENT')
assert hasattr(ph_init, "CONF_UART_CLIENT")
assert ph_init.CONF_UART_CLIENT == "uart_client_id"
assert hasattr(ph_init, 'CONF_LOG_UART_MSG')
assert hasattr(ph_init, "CONF_LOG_UART_MSG")
assert ph_init.CONF_LOG_UART_MSG == "log_uart_msg"
@patch('esphome.components.uart')
@patch('esphome.codegen')
@patch("esphome.components.uart")
@patch("esphome.codegen")
def test_config_schema_structure(self, mock_cg, mock_uart):
"""Test that CONFIG_SCHEMA has the correct structure."""
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
assert hasattr(ph_init, 'CONFIG_SCHEMA')
assert hasattr(ph_init, "CONFIG_SCHEMA")
assert ph_init.CONFIG_SCHEMA is not None
def test_valid_minimal_config(self):
"""Test validation of minimal valid configuration."""
config = {
'id': 'my_heatpump',
'uart_id': 'uart_bus',
"id": "my_heatpump",
"uart_id": "uart_bus",
}
# This would require full ESPHome environment to validate
# Just verify the config structure is reasonable
assert 'id' in config
assert 'uart_id' in config
assert "id" in config
assert "uart_id" in config
def test_valid_full_config(self):
"""Test validation of full configuration with all options."""
config = {
'id': 'my_heatpump',
'uart_id': 'uart_bus',
'uart_client_id': 'uart_client',
'log_uart_msg': True,
'update_interval': '3s',
"id": "my_heatpump",
"uart_id": "uart_bus",
"uart_client_id": "uart_client",
"log_uart_msg": True,
"update_interval": "3s",
}
assert 'id' in config
assert 'uart_id' in config
assert 'uart_client_id' in config
assert 'log_uart_msg' in config
assert 'update_interval' in config
assert "id" in config
assert "uart_id" in config
assert "uart_client_id" in config
assert "log_uart_msg" in config
assert "update_interval" in config
def test_log_uart_msg_default(self):
"""Test that log_uart_msg defaults to False."""
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
# The default is specified in the schema as False
# This test verifies the constant exists
assert ph_init.CONF_LOG_UART_MSG == "log_uart_msg"
@ -157,13 +162,14 @@ class TestPanasonicHeatpumpConfig:
"""Test that component supports multiple instances."""
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
# Verify MULTICONF is True, allowing multiple instances
assert ph_init.MULTICONF is True
@ -171,28 +177,30 @@ class TestPanasonicHeatpumpConfig:
"""Test that UART is listed as a dependency."""
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
assert 'uart' in ph_init.DEPENDENCIES
assert "uart" in ph_init.DEPENDENCIES
def test_namespace_definition(self):
"""Test that component namespace is correctly defined."""
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
# Verify namespace constant exists
assert hasattr(ph_init, 'panasonic_heatpump_ns')
assert hasattr(ph_init, "panasonic_heatpump_ns")
class TestPanasonicHeatpumpPlatforms:
@ -203,12 +211,14 @@ class TestPanasonicHeatpumpPlatforms:
try:
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
from panasonic_heatpump.sensor import __init__ as sensor_init
assert sensor_init is not None
except ImportError:
pytest.skip("Sensor platform not accessible in test environment")
@ -218,12 +228,14 @@ class TestPanasonicHeatpumpPlatforms:
try:
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
from panasonic_heatpump.binary_sensor import __init__ as bs_init
assert bs_init is not None
except ImportError:
pytest.skip("Binary sensor platform not accessible in test environment")
@ -233,12 +245,14 @@ class TestPanasonicHeatpumpPlatforms:
try:
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
from panasonic_heatpump.text_sensor import __init__ as ts_init
assert ts_init is not None
except ImportError:
pytest.skip("Text sensor platform not accessible in test environment")
@ -248,12 +262,14 @@ class TestPanasonicHeatpumpPlatforms:
try:
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
from panasonic_heatpump.number import __init__ as num_init
assert num_init is not None
except ImportError:
pytest.skip("Number platform not accessible in test environment")
@ -263,12 +279,14 @@ class TestPanasonicHeatpumpPlatforms:
try:
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
from panasonic_heatpump.select import __init__ as sel_init
assert sel_init is not None
except ImportError:
pytest.skip("Select platform not accessible in test environment")
@ -278,12 +296,14 @@ class TestPanasonicHeatpumpPlatforms:
try:
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
from panasonic_heatpump.switch import __init__ as sw_init
assert sw_init is not None
except ImportError:
pytest.skip("Switch platform not accessible in test environment")
@ -293,12 +313,14 @@ class TestPanasonicHeatpumpPlatforms:
try:
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
from panasonic_heatpump.climate import __init__ as clim_init
assert clim_init is not None
except ImportError:
pytest.skip("Climate platform not accessible in test environment")
@ -310,35 +332,36 @@ class TestPanasonicHeatpumpCodeGeneration:
@pytest.fixture
def mock_dependencies(self):
"""Mock all ESPHome dependencies."""
with patch('esphome.codegen') as mock_cg, \
patch('esphome.components.uart') as mock_uart, \
patch('esphome.config_validation'):
with patch("esphome.codegen") as mock_cg, patch(
"esphome.components.uart"
) as mock_uart, patch("esphome.config_validation"):
mock_cg.new_Pvariable = MagicMock(return_value=Mock())
mock_cg.register_component = MagicMock()
mock_uart.register_uart_device = MagicMock()
mock_cg.get_variable = MagicMock(return_value=Mock())
mock_cg.add = MagicMock()
yield {
'cg': mock_cg,
'uart': mock_uart,
"cg": mock_cg,
"uart": mock_uart,
}
def test_to_code_function_exists(self):
"""Test that to_code function is defined."""
import sys
import os
components_path = os.path.join(
os.path.dirname(__file__), '..', '..', 'components'
os.path.dirname(__file__), "..", "..", "components"
)
sys.path.insert(0, components_path)
import panasonic_heatpump as ph_init
assert hasattr(ph_init, 'to_code')
assert hasattr(ph_init, "to_code")
assert callable(ph_init.to_code)
if __name__ == '__main__':
pytest.main([__file__, '-v'])
if __name__ == "__main__":
pytest.main([__file__, "-v"])