mirror of
https://github.com/esphome/esphome.git
synced 2026-08-23 06:36:23 +00:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
56dbf680c9 | ||
|
|
f5c92aaefe | ||
|
|
0b66c24e08 | ||
|
|
66424aab68 |
@@ -17,6 +17,9 @@ from esphome.const import (
|
||||
UNIT_OHM,
|
||||
UNIT_PARTS_PER_BILLION,
|
||||
)
|
||||
from esphome.core import ID
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
CONF_RESISTANCE = "resistance"
|
||||
|
||||
@@ -62,7 +65,7 @@ CONFIG_SCHEMA = (
|
||||
FINAL_VALIDATE_SCHEMA = i2c.final_validate_device_schema("ags10", max_frequency="15khz")
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
await i2c.register_i2c_device(var, config)
|
||||
@@ -94,7 +97,12 @@ AGS10_NEW_I2C_ADDRESS_SCHEMA = cv.maybe_simple_value(
|
||||
AGS10_NEW_I2C_ADDRESS_SCHEMA,
|
||||
synchronous=True,
|
||||
)
|
||||
async def ags10newi2caddress_to_code(config, action_id, template_arg, args):
|
||||
async def ags10newi2caddress_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
address = await cg.templatable(config[CONF_ADDRESS], args, cg.uint8)
|
||||
@@ -126,7 +134,12 @@ AGS10_SET_ZERO_POINT_SCHEMA = cv.Schema(
|
||||
AGS10_SET_ZERO_POINT_SCHEMA,
|
||||
synchronous=True,
|
||||
)
|
||||
async def ags10setzeropoint_to_code(config, action_id, template_arg, args):
|
||||
async def ags10setzeropoint_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
mode = await cg.templatable(
|
||||
|
||||
@@ -4,6 +4,9 @@ import esphome.codegen as cg
|
||||
from esphome.components import i2c
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_FREQUENCY, CONF_ID
|
||||
from esphome.core import ID
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
CODEOWNERS = ["@X-Ryl669"]
|
||||
DEPENDENCIES = ["i2c"]
|
||||
@@ -70,7 +73,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
await i2c.register_i2c_device(var, config)
|
||||
@@ -91,7 +94,12 @@ AT581XSettingsAction = at581x_ns.class_("AT581XSettingsAction", automation.Actio
|
||||
),
|
||||
synchronous=True,
|
||||
)
|
||||
async def at581x_reset_to_code(config, action_id, template_arg, args):
|
||||
async def at581x_reset_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
|
||||
@@ -163,7 +171,12 @@ RADAR_SETTINGS_SCHEMA = cv.Schema(
|
||||
RADAR_SETTINGS_SCHEMA,
|
||||
synchronous=True,
|
||||
)
|
||||
async def at581x_settings_to_code(config, action_id, template_arg, args):
|
||||
async def at581x_settings_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ import esphome.codegen as cg
|
||||
from esphome.components import switch
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import DEVICE_CLASS_SWITCH, ICON_WIFI
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .. import CONF_AT581X_ID, AT581XComponent, at581x_ns
|
||||
|
||||
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = switch.switch_schema(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
at581x_component = await cg.get_variable(config[CONF_AT581X_ID])
|
||||
s = await switch.new_switch(config)
|
||||
await cg.register_parented(s, config[CONF_AT581X_ID])
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from esphome import automation
|
||||
import esphome.codegen as cg
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_DATA, CONF_ID, CONF_TRIGGER_ID
|
||||
from esphome.core import CORE, ID
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
CODEOWNERS = ["@mvturnho", "@danielschramm"]
|
||||
IS_PLATFORM_COMPONENT = True
|
||||
@@ -18,7 +21,7 @@ CONF_BIT_RATE = "bit_rate"
|
||||
CONF_ON_FRAME = "on_frame"
|
||||
|
||||
|
||||
def validate_id(config):
|
||||
def validate_id(config: ConfigType) -> ConfigType:
|
||||
if CONF_CAN_ID in config:
|
||||
can_id = config[CONF_CAN_ID]
|
||||
id_ext = config[CONF_USE_EXTENDED_ID]
|
||||
@@ -27,7 +30,7 @@ def validate_id(config):
|
||||
return config
|
||||
|
||||
|
||||
def validate_raw_data(value):
|
||||
def validate_raw_data(value: Any) -> bytes | list:
|
||||
if isinstance(value, str):
|
||||
return value.encode("utf-8")
|
||||
if isinstance(value, list):
|
||||
@@ -71,7 +74,7 @@ CAN_SPEEDS = {
|
||||
}
|
||||
|
||||
|
||||
def get_rate(value):
|
||||
def get_rate(value: str) -> int:
|
||||
match = re.match(r"(\d+)(?:K(\d+)?)?BPS", value, re.IGNORECASE)
|
||||
if not match:
|
||||
raise ValueError(f"Invalid rate format: {value}")
|
||||
@@ -103,7 +106,7 @@ CANBUS_SCHEMA = cv.Schema(
|
||||
CANBUS_SCHEMA.add_extra(validate_id)
|
||||
|
||||
|
||||
async def setup_canbus_core_(var, config):
|
||||
async def setup_canbus_core_(var: MockObj, config: ConfigType) -> None:
|
||||
await cg.register_component(var, config)
|
||||
cg.add(var.set_can_id([config[CONF_CAN_ID]]))
|
||||
cg.add(var.set_use_extended_id([config[CONF_USE_EXTENDED_ID]]))
|
||||
@@ -134,7 +137,7 @@ async def setup_canbus_core_(var, config):
|
||||
)
|
||||
|
||||
|
||||
async def register_canbus(var, config):
|
||||
async def register_canbus(var: MockObj, config: ConfigType) -> None:
|
||||
if not CORE.has_id(config[CONF_ID]):
|
||||
var = cg.new_Pvariable(config[CONF_ID], var)
|
||||
await setup_canbus_core_(var, config)
|
||||
@@ -157,7 +160,12 @@ async def register_canbus(var, config):
|
||||
),
|
||||
synchronous=True,
|
||||
)
|
||||
async def canbus_action_to_code(config, action_id, template_arg, args):
|
||||
async def canbus_action_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_CANBUS_ID])
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ import esphome.codegen as cg
|
||||
from esphome.components import uart
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_ADDRESS, CONF_ID
|
||||
from esphome.types import ConfigType
|
||||
|
||||
CODEOWNERS = ["@s1lvi0"]
|
||||
MULTI_CONF = True
|
||||
@@ -26,7 +27,7 @@ CONFIG_SCHEMA = (
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
await uart.register_uart_device(var, config)
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import esphome.codegen as cg
|
||||
from esphome.components import binary_sensor
|
||||
import esphome.config_validation as cv
|
||||
from esphome.cpp_generator import MockObj
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from . import CONF_BMS_DALY_ID, DalyBmsComponent
|
||||
|
||||
@@ -27,13 +29,13 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def setup_conf(config, key, hub):
|
||||
async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
|
||||
if sensor_config := config.get(key):
|
||||
var = await binary_sensor.new_binary_sensor(sensor_config)
|
||||
cg.add(getattr(hub, f"set_{key}_binary_sensor")(var))
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
|
||||
for key in TYPES:
|
||||
await setup_conf(config, key, hub)
|
||||
|
||||
@@ -23,6 +23,8 @@ from esphome.const import (
|
||||
UNIT_PERCENT,
|
||||
UNIT_VOLT,
|
||||
)
|
||||
from esphome.cpp_generator import MockObj
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from . import CONF_BMS_DALY_ID, DalyBmsComponent
|
||||
|
||||
@@ -222,13 +224,13 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def setup_conf(config, key, hub):
|
||||
async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
|
||||
if sensor_config := config.get(key):
|
||||
sens = await sensor.new_sensor(sensor_config)
|
||||
cg.add(getattr(hub, f"set_{key}_sensor")(sens))
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
|
||||
for key in TYPES:
|
||||
await setup_conf(config, key, hub)
|
||||
|
||||
@@ -2,6 +2,8 @@ import esphome.codegen as cg
|
||||
from esphome.components import text_sensor
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_STATUS
|
||||
from esphome.cpp_generator import MockObj
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from . import CONF_BMS_DALY_ID, DalyBmsComponent
|
||||
|
||||
@@ -23,13 +25,13 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def setup_conf(config, key, hub):
|
||||
async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
|
||||
if sensor_config := config.get(key):
|
||||
sens = await text_sensor.new_text_sensor(sensor_config)
|
||||
cg.add(getattr(hub, f"set_{key}_text_sensor")(sens))
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
|
||||
for key in TYPES:
|
||||
await setup_conf(config, key, hub)
|
||||
|
||||
@@ -38,7 +38,8 @@ from esphome.const import (
|
||||
PLATFORM_NRF52,
|
||||
PlatformFramework,
|
||||
)
|
||||
from esphome.core import CORE
|
||||
from esphome.core import CORE, ID
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
WAKEUP_PINS = {
|
||||
@@ -174,7 +175,7 @@ def validate_config(config: ConfigType) -> ConfigType:
|
||||
return config
|
||||
|
||||
|
||||
def _validate_ex1_wakeup_mode(value):
|
||||
def _validate_ex1_wakeup_mode(value: str) -> str:
|
||||
if value == "ALL_LOW":
|
||||
esp32.only_on_variant(supported=[VARIANT_ESP32], msg_prefix="ALL_LOW")(value)
|
||||
if value == "ANY_LOW":
|
||||
@@ -345,7 +346,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
|
||||
@@ -458,7 +459,12 @@ DEEP_SLEEP_ENTER_SCHEMA = cv.All(
|
||||
DEEP_SLEEP_ENTER_SCHEMA,
|
||||
synchronous=True,
|
||||
)
|
||||
async def deep_sleep_enter_to_code(config, action_id, template_arg, args):
|
||||
async def deep_sleep_enter_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
paren = await cg.get_variable(config[CONF_ID])
|
||||
var = cg.new_Pvariable(action_id, template_arg, paren)
|
||||
if CONF_SLEEP_DURATION in config:
|
||||
@@ -487,7 +493,12 @@ async def deep_sleep_enter_to_code(config, action_id, template_arg, args):
|
||||
automation.maybe_simple_id(DEEP_SLEEP_ACTION_SCHEMA),
|
||||
synchronous=True,
|
||||
)
|
||||
async def deep_sleep_action_to_code(config, action_id, template_arg, args):
|
||||
async def deep_sleep_action_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
return var
|
||||
|
||||
@@ -3,6 +3,9 @@ import esphome.codegen as cg
|
||||
from esphome.components import i2c, time
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_ID
|
||||
from esphome.core import ID
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
CODEOWNERS = ["@badbadc0ffee"]
|
||||
DEPENDENCIES = ["i2c"]
|
||||
@@ -29,7 +32,12 @@ CONFIG_SCHEMA = time.TIME_SCHEMA.extend(
|
||||
),
|
||||
synchronous=True,
|
||||
)
|
||||
async def ds1307_write_time_to_code(config, action_id, template_arg, args):
|
||||
async def ds1307_write_time_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
return var
|
||||
@@ -45,13 +53,18 @@ async def ds1307_write_time_to_code(config, action_id, template_arg, args):
|
||||
),
|
||||
synchronous=True,
|
||||
)
|
||||
async def ds1307_read_time_to_code(config, action_id, template_arg, args):
|
||||
async def ds1307_read_time_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
return var
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
|
||||
await cg.register_component(var, config)
|
||||
|
||||
@@ -38,7 +38,8 @@ from esphome.const import (
|
||||
CONF_SERVICE_UUID,
|
||||
CONF_TRIGGER_ID,
|
||||
)
|
||||
from esphome.core import CORE, CoroPriority, TimePeriod, coroutine_with_priority
|
||||
from esphome.core import CORE, ID, CoroPriority, TimePeriod, coroutine_with_priority
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.enum import StrEnum
|
||||
from esphome.types import ConfigType
|
||||
|
||||
@@ -262,7 +263,7 @@ ESP_BLE_DEVICE_SCHEMA = cv.Schema(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
# Register the loggers this component needs
|
||||
esp32_ble.register_bt_logger(BTLoggers.BLE_SCAN)
|
||||
|
||||
@@ -360,7 +361,7 @@ async def to_code(config):
|
||||
# chance to call register_ble_tracker and register_client before the list is checked
|
||||
# and added to the global defines list.
|
||||
@coroutine_with_priority(CoroPriority.FINAL)
|
||||
async def _add_ble_features():
|
||||
async def _add_ble_features() -> None:
|
||||
# Add feature-specific defines based on what's needed
|
||||
required_features = _get_required_features()
|
||||
# Sensors registered through the neutral ble_device_base path (BLEHub) need
|
||||
@@ -389,8 +390,11 @@ ESP32_BLE_START_SCAN_ACTION_SCHEMA = cv.Schema(
|
||||
synchronous=True,
|
||||
)
|
||||
async def esp32_ble_tracker_start_scan_action_to_code(
|
||||
config, action_id, template_arg, args
|
||||
):
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
paren = await cg.get_variable(config[CONF_ID])
|
||||
var = cg.new_Pvariable(action_id, template_arg, paren)
|
||||
template_ = await cg.templatable(config[CONF_CONTINUOUS], args, cg.bool_)
|
||||
@@ -414,8 +418,11 @@ ESP32_BLE_STOP_SCAN_ACTION_SCHEMA = automation.maybe_simple_id(
|
||||
synchronous=True,
|
||||
)
|
||||
async def esp32_ble_tracker_stop_scan_action_to_code(
|
||||
config, action_id, template_arg, args
|
||||
):
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
return var
|
||||
|
||||
@@ -48,10 +48,12 @@ from esphome.const import (
|
||||
)
|
||||
from esphome.core import (
|
||||
CORE,
|
||||
ID,
|
||||
CoroPriority,
|
||||
TimePeriodMilliseconds,
|
||||
coroutine_with_priority,
|
||||
)
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
@@ -276,7 +278,7 @@ def _validate_spi_interface(config: ConfigType) -> ConfigType:
|
||||
return config
|
||||
|
||||
|
||||
def _validate(config):
|
||||
def _validate(config: ConfigType) -> ConfigType:
|
||||
if CONF_USE_ADDRESS not in config:
|
||||
if CONF_MANUAL_IP in config:
|
||||
use_address = str(config[CONF_MANUAL_IP][CONF_STATIC_IP])
|
||||
@@ -441,7 +443,7 @@ GENERIC_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
def _spi_schema(default_clock: str = "26.67MHz", max_clock: int = int(80e6)):
|
||||
def _spi_schema(default_clock: str = "26.67MHz", max_clock: int = int(80e6)) -> cv.All:
|
||||
return cv.All(
|
||||
BASE_SCHEMA.extend(
|
||||
cv.Schema(
|
||||
@@ -517,7 +519,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
def _final_validate_spi(config):
|
||||
def _final_validate_spi(config: ConfigType) -> None:
|
||||
if not CORE.is_esp32:
|
||||
return # SPI interface validation is ESP32-only
|
||||
if config[CONF_TYPE] not in SPI_ETHERNET_TYPES:
|
||||
@@ -537,7 +539,7 @@ def _final_validate_spi(config):
|
||||
)
|
||||
|
||||
|
||||
def manual_ip(config):
|
||||
def manual_ip(config: ConfigType) -> cg.StructInitializer:
|
||||
return cg.StructInitializer(
|
||||
ManualIP,
|
||||
("static_ip", ip_address_literal(config[CONF_STATIC_IP])),
|
||||
@@ -548,7 +550,7 @@ def manual_ip(config):
|
||||
)
|
||||
|
||||
|
||||
def phy_register(address: int, value: int, page: int):
|
||||
def phy_register(address: int, value: int, page: int) -> cg.StructInitializer:
|
||||
return cg.StructInitializer(
|
||||
PHYRegister,
|
||||
("address", address),
|
||||
@@ -558,7 +560,7 @@ def phy_register(address: int, value: int, page: int):
|
||||
|
||||
|
||||
@coroutine_with_priority(CoroPriority.COMMUNICATION)
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
|
||||
# Apply network priority before register_component (which emits the user's
|
||||
@@ -610,7 +612,7 @@ async def to_code(config):
|
||||
CORE.add_job(final_step)
|
||||
|
||||
|
||||
async def _to_code_esp32(var: cg.Pvariable, config: ConfigType) -> None:
|
||||
async def _to_code_esp32(var: cg.MockObj, config: ConfigType) -> None:
|
||||
from esphome.components.esp32 import (
|
||||
add_idf_component,
|
||||
add_idf_sdkconfig_option,
|
||||
@@ -698,7 +700,7 @@ async def _to_code_esp32(var: cg.Pvariable, config: ConfigType) -> None:
|
||||
add_idf_component(name=component.name, ref=component.version)
|
||||
|
||||
|
||||
async def _to_code_rp2040(var: cg.Pvariable, config: ConfigType) -> None:
|
||||
async def _to_code_rp2040(var: cg.MockObj, config: ConfigType) -> None:
|
||||
cg.add(var.set_clk_pin(config[CONF_CLK_PIN]))
|
||||
cg.add(var.set_miso_pin(config[CONF_MISO_PIN]))
|
||||
cg.add(var.set_mosi_pin(config[CONF_MOSI_PIN]))
|
||||
@@ -793,7 +795,7 @@ FINAL_VALIDATE_SCHEMA = _final_validate
|
||||
|
||||
|
||||
@coroutine_with_priority(CoroPriority.FINAL)
|
||||
async def final_step():
|
||||
async def final_step() -> None:
|
||||
"""Final code generation step to configure optional Ethernet features."""
|
||||
if ip_state_count := CORE.data.get(ETHERNET_IP_STATE_LISTENERS_KEY, 0):
|
||||
cg.add_define("USE_ETHERNET_IP_STATE_LISTENERS")
|
||||
@@ -836,7 +838,12 @@ def _filter_source_files() -> list[str]:
|
||||
FILTER_SOURCE_FILES = _filter_source_files
|
||||
|
||||
|
||||
async def _new_pvariable_to_code(config, id_, template_arg, args):
|
||||
async def _new_pvariable_to_code(
|
||||
config: ConfigType,
|
||||
id_: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
return cg.new_Pvariable(id_, template_arg)
|
||||
|
||||
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from typing import Any
|
||||
|
||||
from esphome import automation
|
||||
from esphome.automation import maybe_simple_id
|
||||
import esphome.codegen as cg
|
||||
@@ -16,6 +18,9 @@ from esphome.const import (
|
||||
UNIT_CELSIUS,
|
||||
UNIT_PERCENT,
|
||||
)
|
||||
from esphome.core import ID
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
DEPENDENCIES = ["i2c"]
|
||||
|
||||
@@ -62,7 +67,7 @@ CONFIG_SCHEMA = (
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
await i2c.register_i2c_device(var, config)
|
||||
@@ -86,7 +91,7 @@ HDC302X_HEATER_POWER_MAP = {
|
||||
}
|
||||
|
||||
|
||||
def heater_power_value(value):
|
||||
def heater_power_value(value: Any) -> cv.Lambda | int:
|
||||
"""Accept enum names or raw uint16 values"""
|
||||
if isinstance(value, cv.Lambda):
|
||||
return value
|
||||
@@ -119,7 +124,12 @@ HDC302X_HEATER_ON_ACTION_SCHEMA = maybe_simple_id(
|
||||
HDC302X_HEATER_ON_ACTION_SCHEMA,
|
||||
synchronous=True,
|
||||
)
|
||||
async def hdc302x_heater_on_to_code(config, action_id, template_arg, args):
|
||||
async def hdc302x_heater_on_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
template_ = await cg.templatable(config[CONF_POWER], args, cg.uint16)
|
||||
@@ -135,7 +145,12 @@ async def hdc302x_heater_on_to_code(config, action_id, template_arg, args):
|
||||
HDC302X_ACTION_SCHEMA,
|
||||
synchronous=True,
|
||||
)
|
||||
async def hdc302x_heater_off_to_code(config, action_id, template_arg, args):
|
||||
async def hdc302x_heater_off_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
return var
|
||||
|
||||
@@ -17,6 +17,9 @@ from esphome.const import (
|
||||
UNIT_EMPTY,
|
||||
UNIT_PERCENT,
|
||||
)
|
||||
from esphome.core import ID
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
DEPENDENCIES = ["i2c"]
|
||||
|
||||
@@ -63,7 +66,7 @@ CONFIG_SCHEMA = (
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
await i2c.register_i2c_device(var, config)
|
||||
@@ -95,7 +98,12 @@ async def to_code(config):
|
||||
),
|
||||
synchronous=True,
|
||||
)
|
||||
async def set_heater_level_to_code(config, action_id, template_arg, args):
|
||||
async def set_heater_level_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
level_ = await cg.templatable(config[CONF_LEVEL], args, cg.uint8)
|
||||
@@ -115,7 +123,12 @@ async def set_heater_level_to_code(config, action_id, template_arg, args):
|
||||
),
|
||||
synchronous=True,
|
||||
)
|
||||
async def set_heater_to_code(config, action_id, template_arg, args):
|
||||
async def set_heater_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
var = cg.new_Pvariable(action_id, template_arg)
|
||||
await cg.register_parented(var, config[CONF_ID])
|
||||
status_ = await cg.templatable(config[CONF_STATUS], args, cg.bool_)
|
||||
|
||||
@@ -60,7 +60,7 @@ FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
await uart.register_uart_device(var, config)
|
||||
|
||||
@@ -2,6 +2,7 @@ import esphome.codegen as cg
|
||||
from esphome.components import binary_sensor
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_TARGET, DEVICE_CLASS_OCCUPANCY
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from . import LD6002BComponent
|
||||
from .const import AREA_COUNT, CONF_LD6002B_ID, MAX_TARGETS
|
||||
@@ -36,7 +37,7 @@ CONFIG_SCHEMA = (
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_LD6002B_ID])
|
||||
|
||||
if target_config := config.get(CONF_TARGET):
|
||||
|
||||
@@ -129,7 +129,7 @@ BUTTON_MAP = {
|
||||
}
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
for key, button_type in BUTTON_MAP.items():
|
||||
if button_config := config.get(key):
|
||||
b = cg.new_Pvariable(button_config[CONF_ID], button_type)
|
||||
|
||||
@@ -136,7 +136,7 @@ def final_validate(config: ConfigType) -> None:
|
||||
FINAL_VALIDATE_SCHEMA = final_validate
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_LD6002B_ID])
|
||||
|
||||
for key, number_type, setter, min_value, max_value, step in (
|
||||
|
||||
@@ -2,6 +2,7 @@ import esphome.codegen as cg
|
||||
from esphome.components import select
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_AREA_ID, CONF_SENSITIVITY, ENTITY_CATEGORY_CONFIG
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .. import LD6002BComponent, ld6002b_ns
|
||||
from ..const import CONF_INSTALLATION_MODE, CONF_LD6002B_ID, CONF_TRIGGER_SPEED
|
||||
@@ -64,7 +65,7 @@ SELECT_MAP = (
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_LD6002B_ID])
|
||||
|
||||
for key, select_type, setter, options in SELECT_MAP:
|
||||
|
||||
@@ -9,6 +9,7 @@ from esphome.const import (
|
||||
STATE_CLASS_MEASUREMENT,
|
||||
UNIT_METER,
|
||||
)
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from . import LD6002BComponent
|
||||
from .const import (
|
||||
@@ -150,7 +151,7 @@ CONFIG_SCHEMA = (
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_LD6002B_ID])
|
||||
|
||||
if target_count_config := config.get(CONF_TARGET_COUNT):
|
||||
|
||||
@@ -2,6 +2,7 @@ import esphome.codegen as cg
|
||||
from esphome.components import switch
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import DEVICE_CLASS_SWITCH, ENTITY_CATEGORY_CONFIG
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .. import LD6002BComponent, ld6002b_ns
|
||||
from ..const import (
|
||||
@@ -46,7 +47,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_LD6002B_ID])
|
||||
|
||||
for key, switch_type, setter in (
|
||||
|
||||
@@ -2,6 +2,7 @@ import esphome.codegen as cg
|
||||
from esphome.components import text_sensor
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import ENTITY_CATEGORY_DIAGNOSTIC
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from . import LD6002BComponent
|
||||
from .const import CONF_LD6002B_ID, CONF_OTA_VERSION, CONF_WORK_MODE
|
||||
@@ -21,7 +22,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_LD6002B_ID])
|
||||
if work_mode_config := config.get(CONF_WORK_MODE):
|
||||
sens = await text_sensor.new_text_sensor(work_mode_config)
|
||||
|
||||
@@ -2,6 +2,7 @@ import esphome.codegen as cg
|
||||
from esphome.components import i2c
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_ID
|
||||
from esphome.types import ConfigType
|
||||
|
||||
DEPENDENCIES = ["i2c"]
|
||||
CODEOWNERS = ["@rnauber"]
|
||||
@@ -26,7 +27,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
).extend(i2c.i2c_device_schema(0x43))
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
await i2c.register_i2c_device(var, config)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import esphome.codegen as cg
|
||||
from esphome.components import binary_sensor
|
||||
import esphome.config_validation as cv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .. import CONF_M5STACK_8ANGLE_ID, M5Stack8AngleComponent, m5stack_8angle_ns
|
||||
|
||||
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_M5STACK_8ANGLE_ID])
|
||||
sens = await binary_sensor.new_binary_sensor(config)
|
||||
cg.add(sens.set_parent(hub))
|
||||
|
||||
@@ -2,6 +2,7 @@ import esphome.codegen as cg
|
||||
from esphome.components import light
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_OUTPUT_ID
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .. import CONF_M5STACK_8ANGLE_ID, M5Stack8AngleComponent, m5stack_8angle_ns
|
||||
|
||||
@@ -21,7 +22,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub = await cg.get_variable(config[CONF_M5STACK_8ANGLE_ID])
|
||||
lights = cg.new_Pvariable(config[CONF_OUTPUT_ID])
|
||||
await light.register_light(lights, config)
|
||||
|
||||
@@ -8,6 +8,7 @@ from esphome.const import (
|
||||
ICON_ROTATE_RIGHT,
|
||||
STATE_CLASS_MEASUREMENT,
|
||||
)
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .. import (
|
||||
CONF_M5STACK_8ANGLE_ID,
|
||||
@@ -55,7 +56,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = await sensor.new_sensor(config)
|
||||
await cg.register_component(var, config)
|
||||
await cg.register_parented(var, config[CONF_M5STACK_8ANGLE_ID])
|
||||
|
||||
@@ -8,8 +8,10 @@ import esphome.codegen as cg
|
||||
from esphome.components import uart
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_ADDRESS, CONF_DISABLE_CRC, CONF_FLOW_CONTROL_PIN, CONF_ID
|
||||
from esphome.cpp_generator import MockObj
|
||||
from esphome.cpp_helpers import gpio_pin_expression
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@@ -84,7 +86,7 @@ CONFIG_SCHEMA = cv.typed_schema(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
cg.add_global(modbus_ns.using)
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
@@ -112,7 +114,9 @@ def _validate_server_address(value: Any) -> int:
|
||||
return address
|
||||
|
||||
|
||||
def modbus_device_schema(default_address, role: Literal["client", "server"] = "client"):
|
||||
def modbus_device_schema(
|
||||
default_address: int | None, role: Literal["client", "server"] = "client"
|
||||
) -> cv.Schema:
|
||||
hub_type = ModbusClient if role == "client" else ModbusServer
|
||||
address_validator = _validate_server_address if role == "server" else cv.hex_uint8_t
|
||||
schema = {
|
||||
@@ -127,14 +131,14 @@ def modbus_device_schema(default_address, role: Literal["client", "server"] = "c
|
||||
|
||||
def final_validate_modbus_device(
|
||||
name: str, *, role: Literal["server", "client"] | None = None
|
||||
):
|
||||
def validate_role(value):
|
||||
) -> cv.Schema:
|
||||
def validate_role(value: str) -> str:
|
||||
assert role in MODBUS_ROLES
|
||||
if value != role:
|
||||
raise cv.Invalid(f"Component {name} requires role to be {role}")
|
||||
return value
|
||||
|
||||
def validate_hub(hub_config):
|
||||
def validate_hub(hub_config: ConfigType) -> ConfigType:
|
||||
hub_schema = {}
|
||||
if role is not None:
|
||||
hub_schema[cv.Required(CONF_ROLE)] = validate_role
|
||||
@@ -147,19 +151,19 @@ def final_validate_modbus_device(
|
||||
)
|
||||
|
||||
|
||||
async def register_modbus_client_device(var, config):
|
||||
async def register_modbus_client_device(var: MockObj, config: ConfigType) -> None:
|
||||
parent = await cg.get_variable(config[CONF_MODBUS_ID])
|
||||
cg.add(var.set_parent(parent))
|
||||
cg.add(var.set_address(config[CONF_ADDRESS]))
|
||||
|
||||
|
||||
async def register_modbus_server_device(var, config):
|
||||
async def register_modbus_server_device(var: MockObj, config: ConfigType) -> None:
|
||||
parent = await cg.get_variable(config[CONF_MODBUS_ID])
|
||||
cg.add(var.set_address(config[CONF_ADDRESS]))
|
||||
cg.add(parent.register_device(var))
|
||||
|
||||
|
||||
async def register_modbus_device(var, config):
|
||||
async def register_modbus_device(var: MockObj, config: ConfigType) -> None:
|
||||
# Remove before 2026.12.0
|
||||
_LOGGER.warning(
|
||||
"'register_modbus_device' is deprecated, use 'register_modbus_client_device' "
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from typing import Any
|
||||
|
||||
from esphome import automation
|
||||
import esphome.codegen as cg
|
||||
from esphome.components.esp32 import (
|
||||
@@ -31,10 +33,12 @@ from esphome.const import (
|
||||
)
|
||||
from esphome.core import (
|
||||
CORE,
|
||||
ID,
|
||||
CoroPriority,
|
||||
TimePeriodMilliseconds,
|
||||
coroutine_with_priority,
|
||||
)
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
@@ -76,7 +80,7 @@ CONF_DEVICE_TYPES = [
|
||||
]
|
||||
|
||||
|
||||
def _validate_txpower(value):
|
||||
def _validate_txpower(value: Any) -> int | float:
|
||||
if CORE.is_esp32:
|
||||
variant = get_esp32_variant()
|
||||
|
||||
@@ -90,7 +94,7 @@ def _validate_txpower(value):
|
||||
return value # Unsupported, fail later with clear error
|
||||
|
||||
|
||||
def set_sdkconfig_options(config):
|
||||
def set_sdkconfig_options(config: ConfigType) -> None:
|
||||
# and expose options for using SPI/UART RCPs
|
||||
add_idf_sdkconfig_option("CONFIG_IEEE802154_ENABLED", True)
|
||||
add_idf_sdkconfig_option("CONFIG_OPENTHREAD_RADIO_NATIVE", True)
|
||||
@@ -180,7 +184,7 @@ def _validate(config: ConfigType) -> ConfigType:
|
||||
return config
|
||||
|
||||
|
||||
def _require_vfs_select(config):
|
||||
def _require_vfs_select(config: ConfigType) -> ConfigType:
|
||||
"""Register VFS select requirement during config validation."""
|
||||
# OpenThread uses esp_vfs_eventfd which requires VFS select support (ESP32 only)
|
||||
if CORE.is_esp32:
|
||||
@@ -188,7 +192,7 @@ def _require_vfs_select(config):
|
||||
return config
|
||||
|
||||
|
||||
def _validate_platform(config):
|
||||
def _validate_platform(config: ConfigType) -> ConfigType:
|
||||
if CORE.using_zephyr:
|
||||
return config
|
||||
return only_on_variant(
|
||||
@@ -203,7 +207,7 @@ def _validate_platform(config):
|
||||
)(config)
|
||||
|
||||
|
||||
def _validate_tlv_hex(value):
|
||||
def _validate_tlv_hex(value: Any) -> str:
|
||||
s = cv.string_strict(value)
|
||||
if len(s) % 2 != 0:
|
||||
raise cv.Invalid("TLV must have an even number of hex characters")
|
||||
@@ -242,7 +246,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
def _final_validate(_):
|
||||
def _final_validate(_: ConfigType) -> None:
|
||||
full_config = fv.full_config.get()
|
||||
network_config = full_config.get("network", {})
|
||||
if not network_config.get(CONF_ENABLE_IPV6, False):
|
||||
@@ -274,7 +278,7 @@ FILTER_SOURCE_FILES = filter_source_files_from_platform(
|
||||
|
||||
|
||||
@coroutine_with_priority(CoroPriority.COMMUNICATION)
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
# Re-enable openthread IDF component (excluded by default)
|
||||
if CORE.is_esp32:
|
||||
include_builtin_idf_component("openthread")
|
||||
@@ -339,7 +343,12 @@ POLL_PERIOD_ACTION_SCHEMA = automation.maybe_conf(
|
||||
POLL_PERIOD_ACTION_SCHEMA,
|
||||
synchronous=True,
|
||||
)
|
||||
async def openthread_poll_period_action_to_code(config, action_id, template_arg, args):
|
||||
async def openthread_poll_period_action_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
paren = await cg.get_variable(config[CONF_ID])
|
||||
var = cg.new_Pvariable(action_id, template_arg, paren)
|
||||
template_ = await cg.templatable(config[CONF_POLL_PERIOD], args, cg.uint32)
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from typing import Any
|
||||
|
||||
from esphome import automation, pins
|
||||
import esphome.codegen as cg
|
||||
from esphome.components import sensor
|
||||
@@ -19,7 +21,9 @@ from esphome.const import (
|
||||
UNIT_PULSES,
|
||||
UNIT_PULSES_PER_MINUTE,
|
||||
)
|
||||
from esphome.core import CORE
|
||||
from esphome.core import CORE, ID
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
CONF_USE_PCNT = "use_pcnt"
|
||||
|
||||
@@ -42,7 +46,7 @@ SetTotalPulsesAction = pulse_counter_ns.class_(
|
||||
)
|
||||
|
||||
|
||||
def validate_internal_filter(value):
|
||||
def validate_internal_filter(value: ConfigType) -> ConfigType:
|
||||
use_pcnt = value.get(CONF_USE_PCNT)
|
||||
if CORE.is_esp8266 and use_pcnt:
|
||||
raise cv.Invalid(
|
||||
@@ -63,7 +67,7 @@ def validate_internal_filter(value):
|
||||
return value
|
||||
|
||||
|
||||
def validate_pulse_counter_pin(value):
|
||||
def validate_pulse_counter_pin(value: Any) -> ConfigType:
|
||||
value = pins.internal_gpio_input_pin_schema(value)
|
||||
if CORE.is_esp8266 and value[CONF_NUMBER] >= 16:
|
||||
raise cv.Invalid(
|
||||
@@ -72,7 +76,7 @@ def validate_pulse_counter_pin(value):
|
||||
return value
|
||||
|
||||
|
||||
def validate_count_mode(value):
|
||||
def validate_count_mode(value: ConfigType) -> ConfigType:
|
||||
rising_edge = value[CONF_RISING_EDGE]
|
||||
falling_edge = value[CONF_FALLING_EDGE]
|
||||
if rising_edge == "DISABLE" and falling_edge == "DISABLE":
|
||||
@@ -126,7 +130,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
use_pcnt = config.get(CONF_USE_PCNT)
|
||||
if CORE.is_esp32 and use_pcnt:
|
||||
include_builtin_idf_component("esp_driver_pcnt")
|
||||
@@ -157,7 +161,12 @@ async def to_code(config):
|
||||
),
|
||||
synchronous=True,
|
||||
)
|
||||
async def set_total_action_to_code(config, action_id, template_arg, args):
|
||||
async def set_total_action_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
paren = await cg.get_variable(config[CONF_ID])
|
||||
var = cg.new_Pvariable(action_id, template_arg, paren)
|
||||
template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32)
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from typing import Any
|
||||
|
||||
from esphome import automation, pins
|
||||
import esphome.codegen as cg
|
||||
from esphome.components import sensor
|
||||
@@ -17,7 +19,9 @@ from esphome.const import (
|
||||
UNIT_PULSES,
|
||||
UNIT_PULSES_PER_MINUTE,
|
||||
)
|
||||
from esphome.core import CORE
|
||||
from esphome.core import CORE, ID, TimePeriodMicroseconds
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
from esphome.types import ConfigType
|
||||
|
||||
CODEOWNERS = ["@stevebaxter", "@cstaahl", "@TrentHouliston"]
|
||||
|
||||
@@ -37,18 +41,18 @@ FILTER_MODES = {
|
||||
SetTotalPulsesAction = pulse_meter_ns.class_("SetTotalPulsesAction", automation.Action)
|
||||
|
||||
|
||||
def validate_internal_filter(value):
|
||||
def validate_internal_filter(value: Any) -> TimePeriodMicroseconds:
|
||||
return cv.positive_time_period_microseconds(value)
|
||||
|
||||
|
||||
def validate_timeout(value):
|
||||
def validate_timeout(value: Any) -> TimePeriodMicroseconds:
|
||||
value = cv.positive_time_period_microseconds(value)
|
||||
if value.total_minutes > 70:
|
||||
raise cv.Invalid("Maximum timeout is 70 minutes")
|
||||
return value
|
||||
|
||||
|
||||
def validate_pulse_meter_pin(value):
|
||||
def validate_pulse_meter_pin(value: Any) -> ConfigType:
|
||||
value = pins.internal_gpio_input_pin_schema(value)
|
||||
if CORE.is_esp8266 and value[CONF_NUMBER] >= 16:
|
||||
raise cv.Invalid(
|
||||
@@ -81,7 +85,7 @@ CONFIG_SCHEMA = sensor.sensor_schema(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = await sensor.new_sensor(config)
|
||||
await cg.register_component(var, config)
|
||||
|
||||
@@ -107,7 +111,12 @@ async def to_code(config):
|
||||
),
|
||||
synchronous=True,
|
||||
)
|
||||
async def set_total_action_to_code(config, action_id, template_arg, args):
|
||||
async def set_total_action_to_code(
|
||||
config: ConfigType,
|
||||
action_id: ID,
|
||||
template_arg: cg.TemplateArguments,
|
||||
args: TemplateArgsType,
|
||||
) -> MockObj:
|
||||
paren = await cg.get_variable(config[CONF_ID])
|
||||
var = cg.new_Pvariable(action_id, template_arg, paren)
|
||||
template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from esphome import external_files, pins
|
||||
import esphome.codegen as cg
|
||||
@@ -66,7 +67,7 @@ KNOWN_FIRMWARE = {
|
||||
}
|
||||
|
||||
|
||||
def parse_firmware_version(value):
|
||||
def parse_firmware_version(value: str) -> tuple[int, int]:
|
||||
match = re.fullmatch(r"(\d+)\.(\d+)", value)
|
||||
if match is None:
|
||||
raise ValueError(f"Not a valid version number {value}")
|
||||
@@ -154,7 +155,7 @@ def _extract_firmware_ref(entry: ConfigType) -> RemoteFile | None:
|
||||
PREFETCH_FILES = external_files.single_stage_prefetch(_extract_firmware_ref)
|
||||
|
||||
|
||||
def validate_firmware(value):
|
||||
def validate_firmware(value: ConfigType) -> ConfigType:
|
||||
config = value.copy()
|
||||
if CONF_URL not in config:
|
||||
try:
|
||||
@@ -167,14 +168,14 @@ def validate_firmware(value):
|
||||
return config
|
||||
|
||||
|
||||
def validate_sha256(value):
|
||||
def validate_sha256(value: Any) -> str:
|
||||
value = cv.string(value)
|
||||
if not re.fullmatch(r"[0-9a-fA-F]{64}", value):
|
||||
raise ValueError(f"Not a valid SHA256 hex string: {value}")
|
||||
return value
|
||||
|
||||
|
||||
def validate_version(value):
|
||||
def validate_version(value: Any) -> str:
|
||||
parse_firmware_version(value)
|
||||
return value
|
||||
|
||||
@@ -231,7 +232,7 @@ FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
|
||||
)
|
||||
|
||||
|
||||
async def to_code(config):
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
fw_hex = get_firmware(config[CONF_FIRMWARE])
|
||||
fw_major, fw_minor = parse_firmware_version(config[CONF_FIRMWARE][CONF_VERSION])
|
||||
|
||||
|
||||
Reference in New Issue
Block a user