Merge branch 'esp8266-native-build-spec' into esp8266-native-ninja-emission

This commit is contained in:
J. Nick Koston
2026-08-22 12:15:40 -05:00
246 changed files with 2473 additions and 663 deletions
+2 -2
View File
@@ -67,7 +67,7 @@ jobs:
with:
python-version: "3.12"
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4.3.0
- name: Determine tag and whether to push
id: tag
@@ -153,7 +153,7 @@ jobs:
with:
python-version: "3.12"
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4.3.0
- name: Log in to the GitHub container registry
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
+2 -2
View File
@@ -123,7 +123,7 @@ jobs:
python-version: "3.12"
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4.3.0
- name: Log in to docker hub
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
@@ -202,7 +202,7 @@ jobs:
merge-multiple: true
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4.3.0
- name: Log in to docker hub
if: matrix.registry == 'dockerhub'
+1 -1
View File
@@ -22,7 +22,7 @@ RUN \
-r /requirements.txt
# Install the ESPHome Device Builder dashboard.
RUN uv pip install --no-cache-dir esphome-device-builder==1.12.1
RUN uv pip install --no-cache-dir esphome-device-builder==1.12.4
RUN \
platformio settings set enable_telemetry No \
+4 -1
View File
@@ -1,3 +1,5 @@
from typing import Any
from esphome import pins
import esphome.codegen as cg
from esphome.components.esp32 import (
@@ -16,6 +18,7 @@ from esphome.components.esp32 import (
import esphome.config_validation as cv
from esphome.const import CONF_ANALOG, CONF_INPUT, CONF_NUMBER, PLATFORM_ESP8266
from esphome.core import CORE
from esphome.types import ConfigType
CODEOWNERS = ["@esphome/core"]
@@ -225,7 +228,7 @@ ESP32_VARIANT_ADC2_PIN_TO_CHANNEL = {
}
def validate_adc_pin(value):
def validate_adc_pin(value: Any) -> ConfigType | str:
if str(value).upper() == "VCC":
if CORE.is_rp2:
return pins.internal_gpio_input_pin_schema(29)
+3 -3
View File
@@ -52,7 +52,7 @@ _attenuation = cv.enum(ATTENUATION_MODES, lower=True)
_sampling_mode = cv.enum(SAMPLING_MODES, lower=True)
def validate_config(config):
def validate_config(config: ConfigType) -> ConfigType:
if config[CONF_RAW] and config.get(CONF_ATTENUATION, None) == "auto":
raise cv.Invalid("Automatic attenuation cannot be used when raw output is set")
@@ -120,7 +120,7 @@ CONFIG_SCHEMA = cv.All(
CONF_ADC_CHANNEL_ID = "adc_channel_id"
def _overlay_io_channels():
def _overlay_io_channels() -> str:
channel_count = CORE.data[CONF_ADC_CHANNEL_ID]
entries = ", ".join(f"<&adc {channel_id}>" for channel_id in range(channel_count))
return f"""
@@ -132,7 +132,7 @@ def _overlay_io_channels():
"""
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 sensor.register_sensor(var, config)
+16 -3
View File
@@ -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(
+8 -1
View File
@@ -6,6 +6,8 @@ from esphome.components.file.image import image_schema, write_image
from esphome.components.image import Image_, validate_settings
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_REPEAT
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CODEOWNERS = ["@syndlex"]
@@ -79,7 +81,12 @@ SET_FRAME_SCHEMA = cv.Schema(
@automation.register_action(
"animation.set_frame", SetFrameAction, SET_FRAME_SCHEMA, synchronous=True
)
async def animation_action_to_code(config, action_id, template_arg, args):
async def animation_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)
+2 -1
View File
@@ -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"]
MULTI_CONF = True
@@ -57,7 +58,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)
+2 -1
View File
@@ -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_DIRECTION, DEVICE_CLASS_MOVING
from esphome.types import ConfigType
from . import APDS9960, CONF_APDS9960_ID
@@ -19,7 +20,7 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_APDS9960_ID])
var = await binary_sensor.new_binary_sensor(config)
func = getattr(hub, f"set_{config[CONF_DIRECTION]}_direction_binary_sensor")
+2 -1
View File
@@ -7,6 +7,7 @@ from esphome.const import (
STATE_CLASS_MEASUREMENT,
UNIT_PERCENT,
)
from esphome.types import ConfigType
from . import APDS9960, CONF_APDS9960_ID
@@ -27,7 +28,7 @@ CONFIG_SCHEMA = sensor.sensor_schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_APDS9960_ID])
var = await sensor.new_sensor(config)
func = getattr(hub, f"set_{config[CONF_TYPE]}_sensor")
+26 -10
View File
@@ -1,5 +1,6 @@
import base64
import logging
from typing import Any
from esphome import automation
from esphome.automation import Condition
@@ -129,7 +130,7 @@ def _register_provisioning_source(config: ConfigType) -> ConfigType:
return config
def validate_encryption_key(value):
def validate_encryption_key(value: Any) -> str:
value = cv.string_strict(value)
try:
decoded = base64.b64decode(value, validate=True)
@@ -217,7 +218,7 @@ def _auto_detect_supports_response(config: ConfigType) -> ConfigType:
return config
def _validate_supports_response(value):
def _validate_supports_response(value: Any) -> str:
"""Validate supports_response after auto-detection has set the value."""
return cv.enum(SUPPORTS_RESPONSE_OPTIONS, lower=True)(value)
@@ -256,7 +257,7 @@ ENCRYPTION_SCHEMA = cv.Schema(
)
def _encryption_schema(config):
def _encryption_schema(config: ConfigType | None) -> ConfigType:
if config is None:
config = {}
return ENCRYPTION_SCHEMA(config)
@@ -393,7 +394,7 @@ async def to_code(config: ConfigType) -> None:
if actions := config.get(CONF_ACTIONS, []):
# Collect all triggers first, then register all at once with initializer_list
triggers: list[cg.Pvariable] = []
triggers: list[cg.MockObj] = []
for conf in actions:
func_args: list[tuple[MockObj, str]] = []
service_template_args: list[MockObj] = [] # User service argument types
@@ -581,7 +582,7 @@ async def homeassistant_service_to_code(
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
):
) -> MockObj:
cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
serv = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, serv, False)
@@ -647,7 +648,7 @@ async def homeassistant_service_to_code(
return var
def validate_homeassistant_event(value):
def validate_homeassistant_event(value: Any) -> str:
value = cv.string(value)
if not value.startswith("esphome."):
raise cv.Invalid(
@@ -676,7 +677,12 @@ HOMEASSISTANT_EVENT_ACTION_SCHEMA = cv.Schema(
HOMEASSISTANT_EVENT_ACTION_SCHEMA,
synchronous=True,
)
async def homeassistant_event_to_code(config, action_id, template_arg, args):
async def homeassistant_event_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
serv = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, serv, True)
@@ -724,7 +730,12 @@ HOMEASSISTANT_TAG_SCANNED_ACTION_SCHEMA = cv.maybe_simple_value(
HOMEASSISTANT_TAG_SCANNED_ACTION_SCHEMA,
synchronous=True,
)
async def homeassistant_tag_scanned_to_code(config, action_id, template_arg, args):
async def homeassistant_tag_scanned_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
serv = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, serv, True)
@@ -740,7 +751,7 @@ CONF_SUCCESS = "success"
CONF_ERROR_MESSAGE = "error_message"
def _validate_api_respond_data(config):
def _validate_api_respond_data(config: ConfigType) -> ConfigType:
"""Set flag during validation so AUTO_LOAD can include json component."""
if CONF_DATA in config:
CORE.data.setdefault(DOMAIN, {})[CONF_CAPTURE_RESPONSE] = True
@@ -824,7 +835,12 @@ API_CONNECTED_CONDITION_SCHEMA = cv.Schema(
@automation.register_condition(
"api.connected", APIConnectedCondition, API_CONNECTED_CONDITION_SCHEMA
)
async def api_connected_to_code(config, condition_id, template_arg, args):
async def api_connected_to_code(
config: ConfigType,
condition_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(condition_id, template_arg)
templ = await cg.templatable(config[CONF_STATE_SUBSCRIPTION_ONLY], args, cg.bool_)
cg.add(var.set_state_subscription_only(templ))
+3 -1
View File
@@ -14,6 +14,8 @@ from esphome.const import (
CONF_TUNE_ANTENNA,
CONF_WATCHDOG_THRESHOLD,
)
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
MULTI_CONF = True
@@ -42,7 +44,7 @@ AS3935_SCHEMA = cv.Schema(
)
async def setup_as3935(var, config):
async def setup_as3935(var: MockObj, config: ConfigType) -> None:
await cg.register_component(var, config)
irq_pin = await cg.gpio_pin_expression(config[CONF_IRQ_PIN])
+2 -1
View File
@@ -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 AS3935, CONF_AS3935_ID
@@ -13,7 +14,7 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema().extend(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_AS3935_ID])
var = await binary_sensor.new_binary_sensor(config)
cg.add(hub.set_thunder_alert_binary_sensor(var))
+2 -1
View File
@@ -9,6 +9,7 @@ from esphome.const import (
STATE_CLASS_MEASUREMENT,
UNIT_KILOMETER,
)
from esphome.types import ConfigType
from . import AS3935, CONF_AS3935_ID
@@ -31,7 +32,7 @@ CONFIG_SCHEMA = cv.Schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_AS3935_ID])
if distance_config := config.get(CONF_DISTANCE):
+16 -3
View File
@@ -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 -1
View File
@@ -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 -1
View File
@@ -86,7 +86,7 @@ AudioFileType detect_audio_file_type(const char *content_type, const char *url)
// Match "audio/ogg" with a codecs parameter containing "opus"
// Valid forms: audio/ogg;codecs=opus, audio/ogg; codecs="opus", etc.
// Plain "audio/ogg" without opus is not matched (almost always Ogg Vorbis)
if (strncasecmp(content_type, "audio/ogg", 9) == 0 && strcasestr(content_type + 9, "opus") != nullptr) {
if (strncasecmp(content_type, "audio/ogg", 9) == 0 && str_contains_ignore_case(content_type + 9, "opus")) {
return AudioFileType::OPUS;
}
#endif
+4 -2
View File
@@ -2,6 +2,8 @@ import esphome.codegen as cg
from esphome.components import ble_client, time
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_RECEIVE_TIMEOUT, CONF_TIME_ID
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
CODEOWNERS = ["@jhansche"]
DEPENDENCIES = ["ble_client"]
@@ -32,12 +34,12 @@ BEDJET_CLIENT_SCHEMA = cv.Schema(
)
async def register_bedjet_child(var, config):
async def register_bedjet_child(var: MockObj, config: ConfigType) -> None:
parent = await cg.get_variable(config[CONF_BEDJET_ID])
cg.add(parent.register_child(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)
await ble_client.register_ble_node(var, config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import climate
import esphome.config_validation as cv
from esphome.const import CONF_HEAT_MODE, CONF_TEMPERATURE_SOURCE
from esphome.types import ConfigType
from .. import BEDJET_CLIENT_SCHEMA, bedjet_ns, register_bedjet_child
@@ -37,7 +38,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await climate.new_climate(config)
await cg.register_component(var, config)
await register_bedjet_child(var, config)
+2 -1
View File
@@ -1,6 +1,7 @@
import esphome.codegen as cg
from esphome.components import fan
import esphome.config_validation as cv
from esphome.types import ConfigType
from .. import BEDJET_CLIENT_SCHEMA, bedjet_ns, register_bedjet_child
@@ -16,7 +17,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await fan.new_fan(config)
await cg.register_component(var, config)
await register_bedjet_child(var, config)
+2 -1
View File
@@ -9,6 +9,7 @@ from esphome.const import (
STATE_CLASS_MEASUREMENT,
UNIT_CELSIUS,
)
from esphome.types import ConfigType
from .. import BEDJET_CLIENT_SCHEMA, bedjet_ns, register_bedjet_child
@@ -38,7 +39,7 @@ CONFIG_SCHEMA = cv.Schema(
).extend(BEDJET_CLIENT_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 register_bedjet_child(var, config)
+2 -1
View File
@@ -3,6 +3,7 @@ from esphome.components import esp32, i2c
from esphome.components.const import CONF_STATE_SAVE_INTERVAL
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_SAMPLE_RATE, CONF_TEMPERATURE_OFFSET, Framework
from esphome.types import ConfigType
CODEOWNERS = ["@trvrnrth"]
DEPENDENCIES = ["i2c"]
@@ -76,7 +77,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)
+4 -2
View File
@@ -29,6 +29,8 @@ from esphome.const import (
UNIT_PARTS_PER_MILLION,
UNIT_PERCENT,
)
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import CONF_BME680_BSEC_ID, SAMPLE_RATE_OPTIONS, BME680BSECComponent
@@ -110,7 +112,7 @@ CONFIG_SCHEMA = cv.Schema(
)
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))
@@ -120,7 +122,7 @@ async def setup_conf(config, key, hub):
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_BME680_BSEC_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_IAQ_ACCURACY
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import CONF_BME680_BSEC_ID, BME680BSECComponent
@@ -21,13 +23,13 @@ CONFIG_SCHEMA = cv.Schema(
)
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_BME680_BSEC_ID])
for key in TYPES:
await setup_conf(config, key, hub)
@@ -3,6 +3,8 @@ from esphome.components import ble_device_base
import esphome.config_validation as cv
from esphome.const import CONF_BINDKEY, CONF_ID, CONF_MAC_ADDRESS
from esphome.core import HexInt
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
CODEOWNERS = ["@nagyrobi"]
AUTO_LOAD = ["ble_device_base"]
@@ -14,7 +16,9 @@ BTHomeMiThermometer = bthome_mithermometer_ns.class_(
)
def bthome_mithermometer_base_schema(extra_schema=None):
def bthome_mithermometer_base_schema(
extra_schema: cv.Schema | dict | None = None,
) -> cv.All:
if extra_schema is None:
extra_schema = {}
return cv.All(
@@ -32,7 +36,7 @@ def bthome_mithermometer_base_schema(extra_schema=None):
)
async def setup_bthome_mithermometer(var, config):
async def setup_bthome_mithermometer(var: MockObj, config: ConfigType) -> None:
await cg.register_component(var, config)
await ble_device_base.register_ble_device(var, config)
cg.add(var.set_address(config[CONF_MAC_ADDRESS].as_hex))
@@ -20,6 +20,7 @@ from esphome.const import (
UNIT_PERCENT,
UNIT_VOLT,
)
from esphome.types import ConfigType
from . import bthome_mithermometer_base_schema, setup_bthome_mithermometer
@@ -67,7 +68,7 @@ CONFIG_SCHEMA = bthome_mithermometer_base_schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await setup_bthome_mithermometer(var, config)
+13 -7
View File
@@ -16,14 +16,15 @@ from esphome.const import (
DEVICE_CLASS_RESTART,
DEVICE_CLASS_UPDATE,
)
from esphome.core import CORE, CoroPriority, coroutine_with_priority
from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
from esphome.core.entity_helpers import (
entity_duplicate_validator,
queue_entity_register,
setup_device_class,
setup_entity,
)
from esphome.cpp_generator import MockObjClass
from esphome.cpp_generator import MockObj, MockObjClass, TemplateArgsType
from esphome.types import ConfigType, SafeExpType
CODEOWNERS = ["@esphome/core"]
IS_PLATFORM_COMPONENT = True
@@ -88,7 +89,7 @@ _CALLBACK_AUTOMATIONS = (
@setup_entity("button")
async def setup_button_core_(var, config):
async def setup_button_core_(var: MockObj, config: ConfigType) -> None:
await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
setup_device_class(config)
@@ -101,7 +102,7 @@ async def setup_button_core_(var, config):
await web_server.add_entity_config(var, web_server_config)
async def register_button(var, config):
async def register_button(var: MockObj, config: ConfigType) -> None:
if not CORE.has_id(config[CONF_ID]):
var = cg.Pvariable(config[CONF_ID], var)
queue_entity_register("button", config)
@@ -109,7 +110,7 @@ async def register_button(var, config):
await setup_button_core_(var, config)
async def new_button(config, *args):
async def new_button(config: ConfigType, *args: SafeExpType) -> MockObj:
var = cg.new_Pvariable(config[CONF_ID], *args)
await register_button(var, config)
return var
@@ -125,11 +126,16 @@ BUTTON_PRESS_SCHEMA = maybe_simple_id(
@automation.register_action(
"button.press", PressAction, BUTTON_PRESS_SCHEMA, synchronous=True
)
async def button_press_to_code(config, action_id, template_arg, args):
async def button_press_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
return cg.new_Pvariable(action_id, template_arg, paren)
@coroutine_with_priority(CoroPriority.CORE)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_global(button_ns.using)
+14 -6
View File
@@ -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])
@@ -6,6 +6,7 @@
#include "esphome/core/string_ref.h"
#include "esphome/components/wifi/wifi_component.h"
#include "captive_index.h"
#include "scan_list.h"
namespace esphome::captive_portal {
@@ -33,8 +34,10 @@ void CaptivePortal::handle_config(AsyncWebServerRequest *request) {
// Invariant: only bounded in-memory work under the lock; the network send
// happens later in request->send()
wifi::ScanResultsLock lock(wifi::global_wifi_component);
for (const auto &scan : wifi::global_wifi_component->get_scan_result()) {
if (scan.get_is_hidden())
const auto &results = wifi::global_wifi_component->get_scan_result();
for (const auto &scan : results) {
bool with_auth = false;
if (!should_show_scan_entry(results, scan, with_auth))
continue;
json_escape_into_buffer(escaped_ssid, scan.get_ssid());
@@ -44,10 +47,10 @@ void CaptivePortal::handle_config(AsyncWebServerRequest *request) {
stream->print(ESPHOME_F("\",\"rssi\":"));
stream->print(scan.get_rssi());
stream->print(ESPHOME_F(",\"lock\":"));
stream->print(scan.get_with_auth());
stream->print(with_auth);
stream->print(ESPHOME_F("}"));
#else
stream->printf(R"(,{"ssid":"%s","rssi":%d,"lock":%d})", escaped_ssid, scan.get_rssi(), scan.get_with_auth());
stream->printf(R"(,{"ssid":"%s","rssi":%d,"lock":%d})", escaped_ssid, scan.get_rssi(), with_auth);
#endif
}
}
@@ -0,0 +1,28 @@
#pragma once
#include <cstdint>
namespace esphome::captive_portal {
// A scan lists every BSSID, so one SSID can appear several times. Returns true for
// the strongest entry per SSID (earliest on ties), never for hidden entries. scan
// must be an element of results. with_auth is written only when returning true and
// is set if any entry with that SSID needs a key. Templated for host tests.
template<typename Results, typename Entry>
bool should_show_scan_entry(const Results &results, const Entry &scan, bool &with_auth) {
if (scan.get_is_hidden())
return false;
const int8_t rssi = scan.get_rssi();
bool any_auth = false;
for (const auto &other : results) {
if (other.get_is_hidden() || !other.ssid_equals(scan))
continue;
// Same array, so address order is index order. scan fails both checks against itself.
if (other.get_rssi() > rssi || (other.get_rssi() == rssi && &other < &scan))
return false;
any_auth |= other.get_with_auth();
}
with_auth = any_auth;
return true;
}
} // namespace esphome::captive_portal
+21 -8
View File
@@ -1,3 +1,5 @@
from typing import Any
from esphome import automation
import esphome.codegen as cg
from esphome.components import mqtt, web_server
@@ -48,13 +50,19 @@ from esphome.const import (
CONF_VISUAL,
CONF_WEB_SERVER,
)
from esphome.core import CORE, CoroPriority, Lambda, coroutine_with_priority
from esphome.core import CORE, ID, CoroPriority, Lambda, coroutine_with_priority
from esphome.core.entity_helpers import (
entity_duplicate_validator,
queue_entity_register,
setup_entity,
)
from esphome.cpp_generator import LambdaExpression, MockObjClass
from esphome.cpp_generator import (
LambdaExpression,
MockObj,
MockObjClass,
TemplateArgsType,
)
from esphome.types import ConfigType, SafeExpType
IS_PLATFORM_COMPONENT = True
@@ -132,7 +140,7 @@ VISUAL_TEMPERATURE_STEP_SCHEMA = cv.Schema(
)
def visual_temperature_step(value):
def visual_temperature_step(value: Any) -> ConfigType:
# Allow defining target/current temperature steps separately
if isinstance(value, dict):
return VISUAL_TEMPERATURE_STEP_SCHEMA(value)
@@ -273,7 +281,7 @@ def climate_schema(
@setup_entity("climate")
async def setup_climate_core_(var, config):
async def setup_climate_core_(var: MockObj, config: ConfigType) -> None:
visual = config.get(CONF_VISUAL, {})
if (min_temp := visual.get(CONF_MIN_TEMPERATURE)) is not None:
cg.add_define("USE_CLIMATE_VISUAL_OVERRIDES")
@@ -443,7 +451,7 @@ async def setup_climate_core_(var, config):
await web_server.add_entity_config(var, web_server_config)
async def register_climate(var, config):
async def register_climate(var: MockObj, config: ConfigType) -> None:
if not CORE.has_id(config[CONF_ID]):
var = cg.Pvariable(config[CONF_ID], var)
queue_entity_register("climate", config)
@@ -451,7 +459,7 @@ async def register_climate(var, config):
await setup_climate_core_(var, config)
async def new_climate(config, *args):
async def new_climate(config: ConfigType, *args: SafeExpType) -> MockObj:
var = cg.new_Pvariable(config[CONF_ID], *args)
await register_climate(var, config)
return var
@@ -485,7 +493,12 @@ CLIMATE_CONTROL_ACTION_SCHEMA = cv.Schema(
CLIMATE_CONTROL_ACTION_SCHEMA,
synchronous=True,
)
async def climate_control_to_code(config, action_id, template_arg, args):
async def climate_control_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
# All configured fields are folded into a single stateless lambda whose
@@ -549,5 +562,5 @@ async def climate_control_to_code(config, action_id, template_arg, args):
@coroutine_with_priority(CoroPriority.CORE)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_global(climate_ns.using)
+7 -4
View File
@@ -1,5 +1,8 @@
from typing import Any
from esphome import codegen as cg, config_validation as cv
from esphome.const import CONF_BLUE, CONF_GREEN, CONF_ID, CONF_RED, CONF_WHITE
from esphome.types import ConfigType
ColorStruct = cg.esphome_ns.struct("Color")
@@ -14,7 +17,7 @@ CONF_WHITE_INT = "white_int"
CONF_HEX = "hex"
def hex_color(value):
def hex_color(value: Any) -> tuple[int, int, int]:
if isinstance(value, int):
value = str(value)
if not isinstance(value, str):
@@ -39,7 +42,7 @@ components = {
}
def validate_color(config):
def validate_color(config: ConfigType) -> ConfigType:
has_components = set(config) & components
has_hex = CONF_HEX in config
if has_hex and has_components:
@@ -68,7 +71,7 @@ CONFIG_SCHEMA = cv.All(
)
def from_rgbw(config):
def from_rgbw(config: ConfigType) -> tuple[int, int, int, int]:
r = 0
if CONF_RED in config:
r = int(config[CONF_RED] * 255)
@@ -96,7 +99,7 @@ def from_rgbw(config):
return (r, g, b, w)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
if CONF_HEX in config:
r, g, b = config[CONF_HEX]
w = 0
+30 -10
View File
@@ -46,7 +46,7 @@ from esphome.core.entity_helpers import (
setup_entity,
)
from esphome.cpp_generator import LambdaExpression, MockObj, MockObjClass
from esphome.types import ConfigType, TemplateArgsType
from esphome.types import ConfigType, SafeExpType, TemplateArgsType
IS_PLATFORM_COMPONENT = True
@@ -162,7 +162,7 @@ _COVER_SCHEMA = (
_COVER_SCHEMA.add_extra(entity_duplicate_validator("cover"))
def _validate_mqtt_state_topics(config):
def _validate_mqtt_state_topics(config: ConfigType) -> ConfigType:
if config.get(CONF_MQTT_JSON_STATE_PAYLOAD):
if CONF_POSITION_STATE_TOPIC in config:
raise cv.Invalid(
@@ -201,7 +201,7 @@ def cover_schema(
@setup_entity("cover")
async def setup_cover_core_(var, config):
async def setup_cover_core_(var: MockObj, config: ConfigType) -> None:
setup_device_class(config)
if CONF_ON_OPEN in config:
@@ -235,7 +235,7 @@ async def setup_cover_core_(var, config):
await web_server.add_entity_config(var, web_server_config)
async def register_cover(var, config):
async def register_cover(var: MockObj, config: ConfigType) -> None:
if not CORE.has_id(config[CONF_ID]):
var = cg.Pvariable(config[CONF_ID], var)
queue_entity_register("cover", config)
@@ -243,7 +243,7 @@ async def register_cover(var, config):
await setup_cover_core_(var, config)
async def new_cover(config, *args):
async def new_cover(config: ConfigType, *args: SafeExpType) -> MockObj:
var = cg.new_Pvariable(config[CONF_ID], *args)
await register_cover(var, config)
return var
@@ -259,7 +259,12 @@ COVER_ACTION_SCHEMA = maybe_simple_id(
@automation.register_action(
"cover.open", OpenAction, COVER_ACTION_SCHEMA, synchronous=True
)
async def cover_open_to_code(config, action_id, template_arg, args):
async def cover_open_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
return cg.new_Pvariable(action_id, template_arg, paren)
@@ -267,7 +272,12 @@ async def cover_open_to_code(config, action_id, template_arg, args):
@automation.register_action(
"cover.close", CloseAction, COVER_ACTION_SCHEMA, synchronous=True
)
async def cover_close_to_code(config, action_id, template_arg, args):
async def cover_close_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
return cg.new_Pvariable(action_id, template_arg, paren)
@@ -275,7 +285,12 @@ async def cover_close_to_code(config, action_id, template_arg, args):
@automation.register_action(
"cover.stop", StopAction, COVER_ACTION_SCHEMA, synchronous=True
)
async def cover_stop_to_code(config, action_id, template_arg, args):
async def cover_stop_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
return cg.new_Pvariable(action_id, template_arg, paren)
@@ -283,7 +298,12 @@ async def cover_stop_to_code(config, action_id, template_arg, args):
@automation.register_action(
"cover.toggle", ToggleAction, COVER_ACTION_SCHEMA, synchronous=True
)
async def cover_toggle_to_code(config, action_id, template_arg, args):
async def cover_toggle_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
return cg.new_Pvariable(action_id, template_arg, paren)
@@ -421,5 +441,5 @@ automation.register_condition(
@coroutine_with_priority(CoroPriority.CORE)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_global(cover_ns.using)
+11 -3
View File
@@ -17,6 +17,9 @@ from esphome.const import (
UNIT_VOLT,
UNIT_WATT,
)
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CODEOWNERS = ["@balrog-kun"]
DEPENDENCIES = ["spi"]
@@ -40,7 +43,7 @@ CONF_VOLTAGE_HPF = "voltage_hpf"
CONF_PULSE_ENERGY = "pulse_energy"
def validate_config(config):
def validate_config(config: ConfigType) -> ConfigType:
current_gain = abs(config[CONF_CURRENT_GAIN]) * (
1.0 if config[CONF_PGA_GAIN] == "10X" else 5.0
)
@@ -105,7 +108,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 spi.register_spi_device(var, config)
@@ -138,6 +141,11 @@ async def to_code(config):
),
synchronous=True,
)
async def restart_action_to_code(config, action_id, template_arg, args):
async def restart_action_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
return cg.new_Pvariable(action_id, template_arg, paren)
+2 -1
View File
@@ -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)
+4 -2
View File
@@ -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)
+4 -2
View File
@@ -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)
+4 -2
View File
@@ -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)
@@ -2,6 +2,7 @@ import base64
from pathlib import Path
import re
import secrets
from typing import Any
import requests
from ruamel.yaml import YAML
@@ -13,6 +14,7 @@ import esphome.config_validation as cv
from esphome.const import CONF_ESPHOME, CONF_PROJECT, CONF_REF, CONF_WIFI
import esphome.final_validate as fv
from esphome.happy_eyeballs import ensure_happy_eyeballs
from esphome.types import ConfigType
from esphome.yaml_util import dump
dashboard_import_ns = cg.esphome_ns.namespace("dashboard_import")
@@ -23,14 +25,14 @@ DEPENDENCIES = ["api"]
CODEOWNERS = ["@esphome/core"]
def validate_import_url(value):
def validate_import_url(value: Any) -> str:
value = cv.string_strict(value)
value = cv.Length(max=255)(value)
validate_source_shorthand(value)
return value
def validate_full_url(config):
def validate_full_url(config: ConfigType) -> ConfigType:
if not config[CONF_IMPORT_FULL_CONFIG]:
return config
source = validate_source_shorthand(config[CONF_PACKAGE_IMPORT_URL])
@@ -55,7 +57,7 @@ CONFIG_SCHEMA = cv.All(
)
def _final_validate(config):
def _final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()[CONF_ESPHOME]
if CONF_PROJECT not in full_config:
raise cv.Invalid(
@@ -73,7 +75,7 @@ wifi:
"""
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_define("USE_DASHBOARD_IMPORT")
url = config[CONF_PACKAGE_IMPORT_URL]
if config[CONF_IMPORT_FULL_CONFIG]:
+2 -1
View File
@@ -12,6 +12,7 @@ from esphome.const import (
PlatformFramework,
)
from esphome.core import CORE
from esphome.types import ConfigType
CODEOWNERS = ["@esphome/core"]
DEPENDENCIES = ["logger"]
@@ -45,7 +46,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
if CORE.using_zephyr:
zephyr_add_prj_conf("HWINFO", True)
# gdb thread support
+2 -1
View File
@@ -21,6 +21,7 @@ from esphome.const import (
UNIT_MILLISECOND,
UNIT_PERCENT,
)
from esphome.types import ConfigType
from . import ( # noqa: F401 pylint: disable=unused-import
CONF_DEBUG_ID,
@@ -111,7 +112,7 @@ CONFIG_SCHEMA = {
}
async def to_code(config):
async def to_code(config: ConfigType) -> None:
debug_component = await cg.get_variable(config[CONF_DEBUG_ID])
if free_conf := config.get(CONF_FREE):
+2 -1
View File
@@ -7,6 +7,7 @@ from esphome.const import (
ICON_CHIP,
ICON_RESTART,
)
from esphome.types import ConfigType
from . import ( # noqa: F401 pylint: disable=unused-import
CONF_DEBUG_ID,
@@ -33,7 +34,7 @@ CONFIG_SCHEMA = cv.Schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
debug_component = await cg.get_variable(config[CONF_DEBUG_ID])
if CONF_DEVICE in config:
+21 -5
View File
@@ -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 = {
@@ -162,6 +163,11 @@ def validate_config(config: ConfigType) -> ConfigType:
"You need to remove the global wakeup_pin_mode and define it per pin"
)
if wakeup_pins:
if CONF_WAKEUP_PIN_MODE in wakeup_pins[0]:
raise cv.Invalid(
"Specify wakeup_pin_mode either at the top level under deep_sleep "
"or under the pin entry, not both"
)
wakeup_pins[0][CONF_WAKEUP_PIN_MODE] = config.pop(CONF_WAKEUP_PIN_MODE)
elif (
isinstance(config.get(CONF_WAKEUP_PIN), list)
@@ -174,7 +180,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 +351,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 +464,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 +498,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
+16 -3
View File
@@ -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)
+4 -3
View File
@@ -3,6 +3,7 @@ import esphome.codegen as cg
from esphome.components import i2c
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_SLEEP_PIN, CONF_TYPE
from esphome.types import ConfigType
CODEOWNERS = ["@tomwellnitz"]
MULTI_CONF = True
@@ -35,7 +36,7 @@ ds248x_ns = cg.esphome_ns.namespace("ds248x")
DS248xComponent = ds248x_ns.class_("DS248xComponent", cg.Component, i2c.I2CDevice)
def _component_schema(*extras):
def _component_schema(*extras: dict) -> cv.Schema:
schema = cv.Schema(
{
cv.GenerateID(): cv.declare_id(DS248xComponent),
@@ -79,11 +80,11 @@ CONFIG_SCHEMA = cv.typed_schema(
)
def get_channel_count(config):
def get_channel_count(config: ConfigType) -> int:
return CHANNEL_COUNTS[config[CONF_TYPE]]
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)
+3 -2
View File
@@ -12,6 +12,7 @@ import esphome.codegen as cg
from esphome.components.one_wire import OneWireBus
import esphome.config_validation as cv
from esphome.const import CONF_CHANNEL, CONF_ID
from esphome.types import ConfigType
from . import CONF_DS248X_ID, DS248xComponent, ds248x_ns, get_channel_count
@@ -29,7 +30,7 @@ CONFIG_SCHEMA = cv.Schema(
).extend(cv.COMPONENT_SCHEMA)
def _final_validate(config):
def _final_validate(config: ConfigType) -> None:
"""Validate that the channel is within the parent's channel count."""
fconf = fv.full_config.get()
path = fconf.get_path_for_id(config[CONF_DS248X_ID])[:-1]
@@ -47,7 +48,7 @@ def _final_validate(config):
FINAL_VALIDATE_SCHEMA = _final_validate
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)
+2 -1
View File
@@ -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, CONF_INVERTED, CONF_RESOLUTION
from esphome.types import ConfigType
CODEOWNERS = ["@ellull"]
@@ -68,7 +69,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)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import output
import esphome.config_validation as cv
from esphome.const import CONF_ID
from esphome.types import ConfigType
from .. import CONF_EMC2101_ID, EMC2101_COMPONENT_SCHEMA, emc2101_ns
@@ -16,7 +17,7 @@ CONFIG_SCHEMA = EMC2101_COMPONENT_SCHEMA.extend(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
paren = await cg.get_variable(config[CONF_EMC2101_ID])
var = cg.new_Pvariable(config[CONF_ID], paren)
await output.register_output(var, config)
@@ -13,6 +13,7 @@ from esphome.const import (
UNIT_PERCENT,
UNIT_REVOLUTIONS_PER_MINUTE,
)
from esphome.types import ConfigType
from .. import CONF_EMC2101_ID, EMC2101_COMPONENT_SCHEMA, emc2101_ns
@@ -53,7 +54,7 @@ CONFIG_SCHEMA = EMC2101_COMPONENT_SCHEMA.extend(
).extend(cv.polling_component_schema("60s"))
async def to_code(config):
async def to_code(config: ConfigType) -> None:
paren = await cg.get_variable(config[CONF_EMC2101_ID])
var = cg.new_Pvariable(config[CONF_ID], paren)
await cg.register_component(var, config)
+7 -3
View File
@@ -11,7 +11,8 @@ from esphome.const import (
CONF_RX_BUFFER_SIZE,
CONF_UART_ID,
)
from esphome.core import CORE
from esphome.core import CORE, ID
from esphome.cpp_generator import MockObj, TemplateArgsType
import esphome.final_validate as fv
from esphome.types import ConfigType
@@ -143,8 +144,11 @@ EMONTX_SEND_COMMAND_ACTION_SCHEMA = cv.Schema(
synchronous=True,
)
async def emontx_send_command_action_to_code(
config: ConfigType, action_id, template_arg, args
) -> None:
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_COMMAND], args, cg.std_string)
+37 -26
View File
@@ -68,6 +68,7 @@ PATTERN_CONFIGS = {
"PULSE": {
CONF_UNIT_OF_MEASUREMENT: UNIT_PULSES,
CONF_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
CONF_STATE_CLASS: STATE_CLASS_TOTAL_INCREASING,
CONF_ACCURACY_DECIMALS: 0,
},
"PF": {
@@ -78,12 +79,13 @@ PATTERN_CONFIGS = {
},
}
# Create a base schema that's flexible for any tag
BASE_SCHEMA = sensor.sensor_schema(
EmonTxSensor,
state_class=STATE_CLASS_MEASUREMENT,
accuracy_decimals=0,
).extend(
# BASE_SCHEMA intentionally omits state_class and accuracy_decimals defaults.
# Passing them to sensor_schema() would register them via cv.Optional(key, default=...),
# making them always present in the validated config dict and preventing
# apply_tag_defaults from overriding them with the correct per-prefix values.
# They are injected by apply_tag_defaults below, after running through
# sensor.validate_state_class() so the value is code-generation-ready.
BASE_SCHEMA = sensor.sensor_schema(EmonTxSensor).extend(
{
cv.GenerateID(CONF_EMONTX_ID): cv.use_id(EmonTx),
cv.Required(CONF_TAG_NAME): cv.string,
@@ -91,34 +93,43 @@ BASE_SCHEMA = sensor.sensor_schema(
)
def _apply_defaults(config: ConfigType, defaults: dict) -> None:
"""Inject defaults into config, skipping keys already set by the user.
state_class values are run through validate_state_class so they are
code-generation-ready, matching what sensor_schema() would normally do."""
for key, value in defaults.items():
if key not in config:
if key == CONF_STATE_CLASS:
value = sensor.validate_state_class(value)
config[key] = value
def apply_tag_defaults(config: ConfigType) -> ConfigType:
"""Apply defaults based on tag prefix if applicable, but don't restrict any tags."""
tag = config[CONF_TAG_NAME]
# Skip if tag is too short
if len(tag) < 2:
return config
if len(tag) >= 2:
tag_upper = tag.upper()
# Check if this tag starts with a known prefix
tag_upper = tag.upper()
for pattern, pattern_config in PATTERN_CONFIGS.items():
if tag_upper.startswith(pattern):
_apply_defaults(config, pattern_config)
return config
for pattern, pattern_config in PATTERN_CONFIGS.items():
if tag_upper.startswith(pattern):
# Apply pattern defaults if not overridden by user
for key, value in pattern_config.items():
if key not in config:
config[key] = value
# Only apply defaults for known prefixes with numeric indices (e.g. E1, V2, T3)
prefix = tag_upper[0]
if prefix in SENSOR_CONFIGS and tag[1:].isdigit():
_apply_defaults(config, SENSOR_CONFIGS[prefix])
return config
# Only apply defaults for known prefixes with numeric indices
prefix = tag_upper[0]
if prefix in SENSOR_CONFIGS and len(tag) > 1 and tag[1:].isdigit():
# Apply defaults for known tag types, but only if not overridden by user
defaults = SENSOR_CONFIGS[prefix]
for key, value in defaults.items():
if key not in config:
config[key] = value
# Fall back to generic defaults for tags with no known prefix
_apply_defaults(
config,
{
CONF_STATE_CLASS: STATE_CLASS_MEASUREMENT,
CONF_ACCURACY_DECIMALS: 0,
},
)
return 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
@@ -135,6 +135,10 @@ void Esp32HostedUpdate::setup() {
// Publish state
this->status_clear_error();
this->publish_state();
// Defer so the automation runs on the main loop after setup, not during App.setup()
if (this->state_ == update::UPDATE_STATE_AVAILABLE && this->update_available_trigger_) {
this->defer([this]() { this->update_available_trigger_->trigger(this->update_info_); });
}
#else
// HTTP mode: check every 10s until network is ready (max 6 attempts)
// Only if update interval is > 1 minute to avoid redundant checks
@@ -185,6 +189,8 @@ void Esp32HostedUpdate::check() {
return;
}
const bool was_available = this->state_ == update::UPDATE_STATE_AVAILABLE;
// Compare versions
if (this->update_info_.latest_version.empty() ||
this->update_info_.latest_version == this->update_info_.current_version) {
@@ -197,6 +203,9 @@ void Esp32HostedUpdate::check() {
this->update_info_.progress = 0.0f;
this->status_clear_error();
this->publish_state();
if (this->state_ == update::UPDATE_STATE_AVAILABLE && !was_available && this->update_available_trigger_) {
this->update_available_trigger_->trigger(this->update_info_);
}
#endif
}
+8 -5
View File
@@ -1,4 +1,6 @@
from collections.abc import Callable, Iterable
import logging
from typing import Any
import esphome.codegen as cg
from esphome.components import esp32
@@ -23,6 +25,7 @@ from esphome.const import (
CONF_VOLTAGE_ATTENUATION,
)
from esphome.core import TimePeriod
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__)
@@ -181,7 +184,7 @@ EFFECTIVE_HIGH_VOLTAGE = {
}
def validate_touch_pad(value):
def validate_touch_pad(value: Any) -> int:
value = gpio.gpio_pin_number_validator(value)
variant = get_esp32_variant()
pads = TOUCH_PADS.get(variant)
@@ -192,7 +195,7 @@ def validate_touch_pad(value):
return pads[value] # Return integer channel ID
def validate_variant_vars(config):
def validate_variant_vars(config: ConfigType) -> ConfigType:
variant = get_esp32_variant()
invalid_vars = set()
if variant == VARIANT_ESP32:
@@ -219,8 +222,8 @@ def validate_variant_vars(config):
return config
def validate_voltage(values):
def validator(value):
def validate_voltage(values: Iterable[str]) -> Callable[[Any], str]:
def validator(value: Any) -> str:
if isinstance(value, float) and value.is_integer():
value = int(value)
value = cv.string(value)
@@ -300,7 +303,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
# New unified touch sensor driver
include_builtin_idf_component("esp_driver_touch_sens")
@@ -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_ID, CONF_PIN, CONF_THRESHOLD
from esphome.types import ConfigType
from . import ESP32TouchComponent, esp32_touch_ns, validate_touch_pad
@@ -24,7 +25,7 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema(ESP32TouchBinarySensor).exten
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_ESP32_TOUCH_ID])
var = cg.new_Pvariable(
config[CONF_ID],
+10 -8
View File
@@ -3,6 +3,7 @@ from pathlib import Path
import platform
import re
import subprocess
from typing import Any
import esphome.codegen as cg
import esphome.config_validation as cv
@@ -31,6 +32,7 @@ from esphome.core import (
from esphome.core.config import BOARD_MAX_LENGTH
from esphome.helpers import IS_MACOS, copy_file_if_changed
from esphome.platformio.toolchain import copy_ccache_script
from esphome.storage_json import StorageJSON
from esphome.types import ConfigType
from .boards import BOARDS, ESP8266_LD_SCRIPTS
@@ -88,7 +90,7 @@ def lambdas_use_scanf_float(config: ConfigType) -> bool:
return False
def set_core_data(config):
def set_core_data(config: ConfigType) -> ConfigType:
CORE.data[KEY_ESP8266] = {}
CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = PLATFORM_ESP8266
CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = "arduino"
@@ -102,7 +104,7 @@ def set_core_data(config):
return config
def get_download_types(storage_json):
def get_download_types(storage_json: StorageJSON) -> list[dict[str, str]]:
"""Binary-download entries for a built ESP8266 firmware.
Used by device-builder (esphome/device-builder), via
@@ -166,7 +168,7 @@ ARDUINO_3_PLATFORM_VERSION = cv.Version(3, 2, 0)
ARDUINO_4_PLATFORM_VERSION = cv.Version(4, 2, 1)
def _arduino_check_versions(value):
def _arduino_check_versions(value: ConfigType) -> ConfigType:
value = value.copy()
lookups = {
"dev": (cv.Version(3, 1, 2), "https://github.com/esp8266/Arduino.git"),
@@ -209,7 +211,7 @@ def _arduino_check_versions(value):
return value
def _parse_platform_version(value):
def _parse_platform_version(value: Any) -> str:
try:
# if platform version is a valid version constraint, prefix the default package
cv.platformio_version_constraint(value)
@@ -287,7 +289,7 @@ def check_rosetta() -> None:
@coroutine_with_priority(CoroPriority.PLATFORM)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add(esp8266_ns.setup_preferences())
cg.add_platformio_option("lib_ldf_mode", "off")
@@ -516,7 +518,7 @@ ESP8266_EXCEPTION_CODES = {
}
def _decode_pc(config, addr):
def _decode_pc(config: ConfigType, addr: str) -> None:
from esphome.platformio import toolchain
idedata = toolchain.get_idedata(config)
@@ -537,7 +539,7 @@ def _decode_pc(config, addr):
_LOGGER.warning("Decoded %s", translation)
def _parse_register(config, regex, line):
def _parse_register(config: ConfigType, regex: re.Pattern[str], line: str) -> None:
match = regex.match(line)
if match is not None:
_decode_pc(config, match.group(1))
@@ -561,7 +563,7 @@ STACKTRACE_BAD_ALLOC_RE = re.compile(
STACKTRACE_ESP8266_BACKTRACE_PC_RE = re.compile(r"4[0-9a-f]{7}")
def process_stacktrace(config, line, backtrace_state):
def process_stacktrace(config: ConfigType, line: str, backtrace_state: bool) -> bool:
line = line.strip()
# ESP8266 Exception type
match = re.match(STACKTRACE_ESP8266_EXCEPTION_TYPE_RE, line)
+9 -6
View File
@@ -1,5 +1,6 @@
from dataclasses import dataclass
import logging
from typing import Any
from esphome import pins
import esphome.codegen as cg
@@ -18,6 +19,8 @@ from esphome.const import (
PLATFORM_ESP8266,
)
from esphome.core import CORE, CoroPriority, coroutine_with_priority
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import boards
from .const import KEY_BOARD, KEY_ESP8266, KEY_PIN_INITIAL_STATES, esp8266_ns
@@ -27,7 +30,7 @@ _LOGGER = logging.getLogger(__name__)
ESP8266GPIOPin = esp8266_ns.class_("ESP8266GPIOPin", cg.InternalGPIOPin)
def _lookup_pin(value):
def _lookup_pin(value: str) -> int:
board = CORE.data[KEY_ESP8266][KEY_BOARD]
board_pins = boards.ESP8266_BOARD_PINS.get(board, {})
@@ -42,7 +45,7 @@ def _lookup_pin(value):
raise cv.Invalid(f"Cannot resolve pin name '{value}' for board {board}.")
def _translate_pin(value):
def _translate_pin(value: Any) -> int:
if isinstance(value, dict) or value is None:
raise cv.Invalid(
"This variable only supports pin numbers, not full pin schemas "
@@ -69,7 +72,7 @@ _ESP_SDIO_PINS = {
}
def validate_gpio_pin(value):
def validate_gpio_pin(value: Any) -> int:
value = _translate_pin(value)
if value < 0 or value > 17:
raise cv.Invalid(f"ESP8266: Invalid pin number: {value}")
@@ -86,7 +89,7 @@ def validate_gpio_pin(value):
return value
def validate_supports(value):
def validate_supports(value: ConfigType) -> ConfigType:
num = value[CONF_NUMBER]
mode = value[CONF_MODE]
is_input = mode[CONF_INPUT]
@@ -160,7 +163,7 @@ class PinInitialState:
@pins.PIN_SCHEMA_REGISTRY.register(PLATFORM_ESP8266, ESP8266_PIN_SCHEMA)
async def esp8266_pin_to_code(config):
async def esp8266_pin_to_code(config: ConfigType) -> MockObj:
var = cg.new_Pvariable(config[CONF_ID])
num = config[CONF_NUMBER]
mode = config[CONF_MODE]
@@ -192,7 +195,7 @@ async def esp8266_pin_to_code(config):
@coroutine_with_priority(CoroPriority.WORKAROUNDS)
async def add_pin_initial_states_array():
async def add_pin_initial_states_array() -> None:
# Add includes at the very end, so that they override everything
initial_states: list[PinInitialState] = CORE.data[KEY_ESP8266][
KEY_PIN_INITIAL_STATES
+11 -3
View File
@@ -4,11 +4,14 @@ from esphome.components import output
from esphome.components.esp8266.const import require_waveform
import esphome.config_validation as cv
from esphome.const import CONF_FREQUENCY, CONF_ID, CONF_NUMBER, CONF_PIN
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
DEPENDENCIES = ["esp8266"]
def valid_pwm_pin(value):
def valid_pwm_pin(value: ConfigType) -> ConfigType:
num = value[CONF_NUMBER]
cv.one_of(0, 1, 2, 3, 4, 5, 9, 10, 12, 13, 14, 15, 16)(num)
return value
@@ -35,7 +38,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config) -> None:
async def to_code(config: ConfigType) -> None:
require_waveform()
var = cg.new_Pvariable(config[CONF_ID])
@@ -59,7 +62,12 @@ async def to_code(config) -> None:
),
synchronous=True,
)
async def esp8266_set_frequency_to_code(config, action_id, template_arg, args):
async def esp8266_set_frequency_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_FREQUENCY], args, cg.float_)
@@ -129,14 +129,17 @@ void on_data_received(const esp_now_recv_info_t *info, const uint8_t *data, int
ESPNowComponent::ESPNowComponent() { global_esp_now = this; }
void ESPNowComponent::dump_config() {
uint32_t version = 0;
esp_now_get_version(&version);
ESP_LOGCONFIG(TAG, "espnow:");
if (this->is_disabled()) {
ESP_LOGCONFIG(TAG, " Disabled");
// Only report driver details once enabled; with enable_on_boot: false the
// Wi-Fi driver is not initialized yet and esp_now_get_version() would crash,
// and after a failed enable_() the values would be meaningless.
if (this->state_ != ESPNOW_STATE_ENABLED) {
// OFF here means enable_() failed; the core logs the FAILED marker separately
ESP_LOGCONFIG(TAG, " %s", this->is_disabled() ? LOG_STR_LITERAL("Disabled") : LOG_STR_LITERAL("Not enabled"));
return;
}
uint32_t version = 0;
esp_now_get_version(&version);
char own_addr_buf[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
format_mac_addr_upper(this->own_address_, own_addr_buf);
ESP_LOGCONFIG(TAG,
+17 -10
View File
@@ -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")
@@ -845,7 +847,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)
+4 -3
View File
@@ -12,6 +12,7 @@ from esphome.const import (
)
from esphome.core import CORE
from esphome.final_validate import full_config
from esphome.types import ConfigType
CODEOWNERS = ["@anatoly-savchenkov"]
@@ -23,7 +24,7 @@ CONF_RESETS_REQUIRED = "resets_required"
CONF_ON_INCREMENT = "on_increment"
def _validate(config):
def _validate(config: ConfigType) -> ConfigType:
if CONF_RESETS_REQUIRED in config:
return cv.only_on(
[
@@ -60,7 +61,7 @@ CONFIG_SCHEMA = cv.All(
)
def _final_validate(config) -> None:
def _final_validate(config: ConfigType) -> None:
if CORE.is_esp8266 and CONF_RESETS_REQUIRED in config:
fconfig = full_config.get()
if not fconfig.get_config_for_path([KEY_ESP8266, CONF_RESTORE_FROM_FLASH]):
@@ -81,7 +82,7 @@ _CALLBACK_AUTOMATIONS = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
if reset_count := config.get(CONF_RESETS_REQUIRED):
var = cg.new_Pvariable(
config[CONF_ID],
@@ -7,6 +7,7 @@ from esphome.const import (
ENTITY_CATEGORY_CONFIG,
ICON_RESTART_ALERT,
)
from esphome.types import ConfigType
from .. import factory_reset_ns
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = button.button_schema(
).extend(cv.COMPONENT_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 button.register_button(var, config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import switch
import esphome.config_validation as cv
from esphome.const import ENTITY_CATEGORY_CONFIG, ICON_RESTART_ALERT
from esphome.types import ConfigType
from .. import factory_reset_ns
@@ -17,6 +18,6 @@ CONFIG_SCHEMA = switch.switch_schema(
).extend(cv.COMPONENT_SCHEMA)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await switch.new_switch(config)
await cg.register_component(var, config)
+10 -7
View File
@@ -5,6 +5,7 @@ import io
import logging
from pathlib import Path
import re
from typing import Any
from PIL import Image, UnidentifiedImageError
@@ -75,12 +76,12 @@ def compute_local_image_path(value: str | ConfigType) -> Path:
return external_files.compute_local_file_path(DOMAIN, url)
def local_path(value):
def local_path(value: str | ConfigType) -> str:
value = value[CONF_PATH] if isinstance(value, dict) else value
return str(CORE.relative_config_path(value))
def download_file(url, path):
def download_file(url: str, path: Path) -> str:
# The shared NETWORK_TIMEOUT applies; a per-caller timeout would be
# silently ignored on a per-run memo hit anyway (memos key by path).
external_files.download_content(url, path)
@@ -98,7 +99,7 @@ def download_gh_svg(value: str | ConfigType, source: str) -> str:
return download_file(url, path)
def download_image(value):
def download_image(value: str | ConfigType) -> str:
value = value[CONF_URL] if isinstance(value, dict) else value
return download_file(value, compute_local_image_path(value))
@@ -146,7 +147,7 @@ def _extract_entry_ref(entry: ConfigType) -> RemoteFile | None:
PREFETCH_FILES = external_files.single_stage_prefetch(_extract_entry_ref)
def validate_file_shorthand(value):
def validate_file_shorthand(value: Any) -> str:
value = cv.string_strict(value)
if (remote := _parse_remote_shorthand(value)) is not None:
return download_file(remote.url, remote.path)
@@ -163,8 +164,8 @@ LOCAL_SCHEMA = cv.All(
)
def mdi_schema(source):
def validate_mdi(value):
def mdi_schema(source: str) -> cv.All:
def validate_mdi(value: ConfigType) -> str:
return download_gh_svg(value, source)
return cv.All(
@@ -259,7 +260,9 @@ async def new_image(config: ConfigType) -> MockObj:
return var
async def write_image(config, all_frames=False):
async def write_image(
config: ConfigType, all_frames: bool = False
) -> tuple[MockObj, int, int, MockObj, MockObj, int]:
path = Path(config[CONF_FILE])
if not path.is_file():
raise core.EsphomeError(f"Could not load image file {path}")
+2 -1
View File
@@ -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
CODEOWNERS = ["@Szewcson"]
@@ -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 i2c.register_i2c_device(var, config)
+2 -1
View File
@@ -7,6 +7,7 @@ from esphome.const import (
ENTITY_CATEGORY_DIAGNOSTIC,
ICON_VIBRATE,
)
from esphome.types import ConfigType
from . import CONF_GDK101_ID, GDK101Component
@@ -24,7 +25,7 @@ CONFIG_SCHEMA = cv.Schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_GDK101_ID])
var = await binary_sensor.new_binary_sensor(config[CONF_VIBRATIONS])
cg.add(hub.set_vibration_binary_sensor(var))
+2 -1
View File
@@ -15,6 +15,7 @@ from esphome.const import (
UNIT_MICROSILVERTS_PER_HOUR,
UNIT_SECOND,
)
from esphome.types import ConfigType
from . import CONF_GDK101_ID, GDK101Component
@@ -59,7 +60,7 @@ CONFIG_SCHEMA = cv.Schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_GDK101_ID])
if radiation_dose_per_1m := config.get(CONF_RADIATION_DOSE_PER_1M):
+2 -1
View File
@@ -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 CONF_VERSION, ENTITY_CATEGORY_DIAGNOSTIC, ICON_CHIP
from esphome.types import ConfigType
from . import CONF_GDK101_ID, GDK101Component
@@ -17,7 +18,7 @@ CONFIG_SCHEMA = cv.Schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_GDK101_ID])
var = await text_sensor.new_text_sensor(config[CONF_VERSION])
cg.add(hub.set_fw_version_text_sensor(var))
+9 -3
View File
@@ -8,7 +8,8 @@ from esphome.const import (
CONF_TYPE,
CONF_VALUE,
)
from esphome.core import CoroPriority, coroutine_with_priority
from esphome.core import ID, CoroPriority, coroutine_with_priority
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CODEOWNERS = ["@esphome/core"]
@@ -62,7 +63,7 @@ CONFIG_SCHEMA = _globals_schema
# Run with low priority so that namespaces are registered first
@coroutine_with_priority(CoroPriority.LATE)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
type_ = cg.RawExpression(config[CONF_TYPE])
restore = config[CONF_RESTORE_VALUE]
@@ -104,7 +105,12 @@ async def to_code(config):
),
synchronous=True,
)
async def globals_set_to_code(config, action_id, template_arg, args):
async def globals_set_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
full_id, paren = await cg.get_variable_with_full_id(config[CONF_ID])
template_arg = cg.TemplateArguments(full_id.type, *template_arg)
var = cg.new_Pvariable(action_id, template_arg, paren)
@@ -12,6 +12,7 @@ from esphome.const import (
CONF_PIN,
)
from esphome.core import CORE
from esphome.types import ConfigType
from .. import gpio_ns
@@ -68,7 +69,7 @@ def _pin_shared_only_with_deep_sleep(pin_num: int) -> bool:
return any(path and path[0] == "deep_sleep" for path, _, _ in pin_users)
def _final_validate(config) -> None:
def _final_validate(config: ConfigType) -> None:
use_interrupt = config[CONF_USE_INTERRUPT]
if not use_interrupt:
return
@@ -124,7 +125,7 @@ def _final_validate(config) -> None:
FINAL_VALIDATE_SCHEMA = _final_validate
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await binary_sensor.new_binary_sensor(config)
await cg.register_component(var, config)
+2 -1
View File
@@ -3,6 +3,7 @@ import esphome.codegen as cg
from esphome.components.one_wire import OneWireBus
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_PIN
from esphome.types import ConfigType
from .. import gpio_ns
@@ -18,7 +19,7 @@ CONFIG_SCHEMA = cv.Schema(
).extend(cv.COMPONENT_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)
+2 -1
View File
@@ -3,6 +3,7 @@ import esphome.codegen as cg
from esphome.components import output
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_PIN
from esphome.types import ConfigType
from .. import gpio_ns
@@ -16,7 +17,7 @@ CONFIG_SCHEMA = output.BINARY_OUTPUT_SCHEMA.extend(
).extend(cv.COMPONENT_SCHEMA)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await output.register_output(var, config)
await cg.register_component(var, config)
+2 -1
View File
@@ -3,6 +3,7 @@ import esphome.codegen as cg
from esphome.components import switch
import esphome.config_validation as cv
from esphome.const import CONF_INTERLOCK, CONF_PIN
from esphome.types import ConfigType
from .. import gpio_ns
@@ -24,7 +25,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await switch.new_switch(config)
await cg.register_component(var, config)
+6 -3
View File
@@ -29,6 +29,7 @@ from esphome.const import (
CONF_X_GRID,
CONF_Y_GRID,
)
from esphome.types import ConfigType
CODEOWNERS = ["@synco"]
@@ -115,7 +116,9 @@ GRAPH_SCHEMA = cv.Schema(
)
def _relocate_fields_to_subfolder(config, subfolder, subschema):
def _relocate_fields_to_subfolder(
config: ConfigType, subfolder: str, subschema: cv.Schema
) -> ConfigType:
fields = [k.schema for k in subschema.schema]
fields.remove(CONF_ID)
if subfolder in config:
@@ -138,7 +141,7 @@ def _relocate_fields_to_subfolder(config, subfolder, subschema):
return config
def _relocate_trace(config):
def _relocate_trace(config: ConfigType) -> ConfigType:
return _relocate_fields_to_subfolder(config, CONF_TRACES, GRAPH_TRACE_SCHEMA)
@@ -148,7 +151,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
cg.add(var.set_duration(config[CONF_DURATION]))
cg.add(var.set_width(config[CONF_WIDTH]))
+10 -2
View File
@@ -13,6 +13,9 @@ from esphome.const import (
CONF_PRESET_MODES,
CONF_SPEED_COUNT,
)
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
from .. import hbridge_ns
@@ -54,12 +57,17 @@ CONFIG_SCHEMA = (
maybe_simple_id({cv.GenerateID(): cv.use_id(HBridgeFan)}),
synchronous=True,
)
async def fan_hbridge_brake_to_code(config, action_id, template_arg, args):
async def fan_hbridge_brake_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
return cg.new_Pvariable(action_id, template_arg, paren)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await fan.new_fan(
config,
config[CONF_SPEED_COUNT],
+2 -1
View File
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import light, output
import esphome.config_validation as cv
from esphome.const import CONF_OUTPUT_ID, CONF_PIN_A, CONF_PIN_B, CONF_UPDATE_INTERVAL
from esphome.types import ConfigType
from .. import hbridge_ns
@@ -21,7 +22,7 @@ CONFIG_SCHEMA = light.RGB_LIGHT_SCHEMA.extend(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_OUTPUT_ID])
cg.add(var.set_update_interval(config.pop(CONF_UPDATE_INTERVAL)))
await cg.register_component(var, config)
@@ -3,6 +3,7 @@ import esphome.codegen as cg
from esphome.components import switch
import esphome.config_validation as cv
from esphome.const import CONF_OPTIMISTIC, CONF_PULSE_LENGTH, CONF_WAIT_TIME
from esphome.types import ConfigType
from .. import hbridge_ns
@@ -30,7 +31,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await switch.new_switch(config)
await cg.register_component(var, config)
+10 -2
View File
@@ -12,6 +12,9 @@ from esphome.const import (
STATE_CLASS_MEASUREMENT,
UNIT_PARTS_PER_MILLION,
)
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
DEPENDENCIES = ["uart"]
@@ -47,7 +50,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)
@@ -73,7 +76,12 @@ CALIBRATION_ACTION_SCHEMA = cv.Schema(
CALIBRATION_ACTION_SCHEMA,
synchronous=True,
)
async def hc8_calibration_to_code(config, action_id, template_arg, args):
async def hc8_calibration_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_BASELINE], args, cg.uint16)
+19 -4
View File
@@ -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
+11 -4
View File
@@ -1,3 +1,6 @@
from collections.abc import Callable
from typing import Any
import esphome.codegen as cg
from esphome.components import i2c, sensor
import esphome.config_validation as cv
@@ -17,6 +20,8 @@ from esphome.const import (
UNIT_DEGREES,
UNIT_MICROTESLA,
)
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
DEPENDENCIES = ["i2c"]
@@ -59,14 +64,16 @@ HMC5883L_RANGES = {
}
def validate_enum(enum_values, units=None, int=True):
def validate_enum(
enum_values: dict[Any, Any], units: str | list[str] | None = None, int: bool = True
) -> Callable[[Any], Any]:
_units = []
if units is not None:
_units = units if isinstance(units, list) else [units]
_units = [str(x) for x in _units]
enum_bound = cv.enum(enum_values, int=int)
def validate_enum_bound(value):
def validate_enum_bound(value: Any) -> Any:
value = cv.string(value)
for unit in _units:
if value.endswith(unit):
@@ -112,7 +119,7 @@ CONFIG_SCHEMA = (
)
def auto_data_rate(config):
def auto_data_rate(config: ConfigType) -> MockObj:
interval_msec = config[CONF_UPDATE_INTERVAL].total_milliseconds
interval_hz = 1000.0 / interval_msec
for datarate in sorted(HMC5883LDatarates.keys()):
@@ -121,7 +128,7 @@ def auto_data_rate(config):
return HMC5883LDatarates[75]
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)
+9 -3
View File
@@ -1,13 +1,19 @@
from collections.abc import Callable, Iterable
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.const import CONF_ATTRIBUTE, CONF_ENTITY_ID, CONF_INTERNAL
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
CODEOWNERS = ["@OttoWinter", "@esphome/core"]
homeassistant_ns = cg.esphome_ns.namespace("homeassistant")
def validate_entity_domain(platform, supported_domains):
def validator(config):
def validate_entity_domain(
platform: str, supported_domains: Iterable[str]
) -> Callable[[ConfigType], ConfigType]:
def validator(config: ConfigType) -> ConfigType:
domain = config[CONF_ENTITY_ID].split(".", 1)[0]
if domain not in supported_domains:
raise cv.Invalid(
@@ -34,7 +40,7 @@ HOME_ASSISTANT_IMPORT_CONTROL_SCHEMA = cv.Schema(
)
def setup_home_assistant_entity(var, config):
def setup_home_assistant_entity(var: MockObj, config: ConfigType) -> None:
cg.add(var.set_entity_id(config[CONF_ENTITY_ID]))
if CONF_ATTRIBUTE in config:
cg.add(var.set_attribute(config[CONF_ATTRIBUTE]))
@@ -1,5 +1,6 @@
import esphome.codegen as cg
from esphome.components import binary_sensor
from esphome.types import ConfigType
from .. import (
HOME_ASSISTANT_IMPORT_SCHEMA,
@@ -18,7 +19,7 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema(HomeassistantBinarySensor).ex
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await binary_sensor.new_binary_sensor(config)
await cg.register_component(var, config)
setup_home_assistant_entity(var, config)
@@ -1,6 +1,7 @@
import esphome.codegen as cg
from esphome.components import number
import esphome.config_validation as cv
from esphome.types import ConfigType
from .. import (
HOME_ASSISTANT_IMPORT_CONTROL_SCHEMA,
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
var = await number.new_number(
config,
@@ -1,5 +1,6 @@
import esphome.codegen as cg
from esphome.components import sensor
from esphome.types import ConfigType
from .. import (
HOME_ASSISTANT_IMPORT_SCHEMA,
@@ -18,7 +19,7 @@ CONFIG_SCHEMA = sensor.sensor_schema(HomeassistantSensor, accuracy_decimals=1).e
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await sensor.new_sensor(config)
await cg.register_component(var, config)
setup_home_assistant_entity(var, config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import switch
import esphome.config_validation as cv
from esphome.const import CONF_ID
from esphome.types import ConfigType
from .. import (
HOME_ASSISTANT_IMPORT_CONTROL_SCHEMA,
@@ -36,7 +37,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
@@ -1,5 +1,6 @@
import esphome.codegen as cg
from esphome.components import text_sensor
from esphome.types import ConfigType
from .. import (
HOME_ASSISTANT_IMPORT_SCHEMA,
@@ -18,7 +19,7 @@ CONFIG_SCHEMA = text_sensor.text_sensor_schema(HomeassistantTextSensor).extend(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await text_sensor.new_text_sensor(config)
await cg.register_component(var, config)
setup_home_assistant_entity(var, config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import time as time_
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_TIMEZONE
from esphome.types import ConfigType
from .. import homeassistant_ns
@@ -16,7 +17,7 @@ CONFIG_SCHEMA = time_.TIME_SCHEMA.extend(
).extend(cv.COMPONENT_SCHEMA)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await time_.register_time(var, config)
await cg.register_component(var, config)
+3 -2
View File
@@ -11,6 +11,7 @@ from esphome.const import (
)
from esphome.core import CORE
from esphome.platformio.toolchain import copy_ccache_script
from esphome.types import ConfigType
from .const import KEY_HOST
@@ -22,7 +23,7 @@ AUTO_LOAD = ["network", "preferences"]
IS_TARGET_PLATFORM = True
def set_core_data(config):
def set_core_data(config: ConfigType) -> ConfigType:
CORE.data[KEY_HOST] = {}
CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = PLATFORM_HOST
CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = "host"
@@ -41,7 +42,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_build_flag("-DUSE_HOST")
cg.add_define("USE_NATIVE_64BIT_TIME")
# The prefs file finds stored preferences by key, so key migration is possible
+6 -3
View File
@@ -1,4 +1,5 @@
import logging
from typing import Any
from esphome import pins
import esphome.codegen as cg
@@ -14,6 +15,8 @@ from esphome.const import (
CONF_PULLDOWN,
CONF_PULLUP,
)
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from .const import host_ns
@@ -22,7 +25,7 @@ _LOGGER = logging.getLogger(__name__)
HostGPIOPin = host_ns.class_("HostGPIOPin", cg.InternalGPIOPin)
def _translate_pin(value):
def _translate_pin(value: Any) -> int | str:
if isinstance(value, dict) or value is None:
raise cv.Invalid(
"This variable only supports pin numbers, not full pin schemas "
@@ -41,7 +44,7 @@ def _translate_pin(value):
return value
def validate_gpio_pin(value):
def validate_gpio_pin(value: Any) -> int | str:
return _translate_pin(value)
@@ -53,7 +56,7 @@ HOST_PIN_SCHEMA = pins.gpio_base_schema(
@pins.PIN_SCHEMA_REGISTRY.register("host", HOST_PIN_SCHEMA)
async def host_pin_to_code(config):
async def host_pin_to_code(config: ConfigType) -> MockObj:
var = cg.new_Pvariable(config[CONF_ID])
num = config[CONF_NUMBER]
cg.add(var.set_pin(num))
+2 -1
View File
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import time as time_
import esphome.config_validation as cv
from esphome.const import CONF_ID
from esphome.types import ConfigType
CODEOWNERS = ["@clydebarrow"]
@@ -14,7 +15,7 @@ CONFIG_SCHEMA = time_.TIME_SCHEMA.extend(
).extend(cv.COMPONENT_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 time_.register_time(var, config)
+16 -3
View File
@@ -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_)
+17 -15
View File
@@ -1,6 +1,7 @@
import logging
import re
import sys
from typing import Any
from esphome import pins
import esphome.codegen as cg
@@ -52,9 +53,10 @@ from esphome.const import (
PLATFORM_RP2,
PlatformFramework,
)
from esphome.core import CORE, CoroPriority, coroutine_with_priority
from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
from esphome.cpp_generator import MockObj
import esphome.final_validate as fv
from esphome.types import ConfigType
LOGGER = logging.getLogger(__name__)
CODEOWNERS = ["@esphome/core"]
@@ -96,13 +98,13 @@ CONF_SCL_PULLUP_ENABLED = "scl_pullup_enabled"
MULTI_CONF = True
def validate_device(value):
def validate_device(value: str) -> str:
if not re.match(r"^/(?:[^/]+/)*[^/]+$", value):
raise cv.Invalid("Device must be an absolute device path (e.g., /dev/i2c-0)")
return value
def _bus_declare_type(value):
def _bus_declare_type(value: Any) -> ID:
if CORE.is_esp32:
return cv.declare_id(IDFI2CBus)(value)
if CORE.using_arduino:
@@ -114,7 +116,7 @@ def _bus_declare_type(value):
raise NotImplementedError
def _rp2040_i2c_controller(pin):
def _rp2040_i2c_controller(pin: int) -> int:
"""Return the I2C controller number (0 or 1) for a given RP2040/RP2350 GPIO pin.
See RP2040 datasheet Table 2 (section 1.4.3, "GPIO Functions"):
@@ -125,7 +127,7 @@ def _rp2040_i2c_controller(pin):
return (pin // 2) % 2
def validate_config(config):
def validate_config(config: ConfigType) -> ConfigType:
if CORE.is_esp32:
return cv.require_framework_version(
esp_idf=cv.Version(5, 4, 2), esp32_arduino=cv.Version(3, 2, 1)
@@ -142,7 +144,7 @@ def validate_config(config):
return config
def validate_host_config(config):
def validate_host_config(config: ConfigType) -> ConfigType:
if CORE.is_host:
# Host I2C is currently only supported on Linux
if not sys.platform.lower().startswith("linux"):
@@ -229,7 +231,7 @@ CONFIG_SCHEMA = cv.All(
)
def _final_validate(config):
def _final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()[CONF_I2C]
if CORE.using_zephyr and len(full_config) > 1:
raise cv.Invalid("Second i2c is not implemented on Zephyr yet")
@@ -281,7 +283,7 @@ FINAL_VALIDATE_SCHEMA = _final_validate
@coroutine_with_priority(CoroPriority.BUS)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_global(i2c_ns.using)
cg.add_define("USE_I2C")
if CORE.is_esp32:
@@ -358,7 +360,7 @@ async def to_code(config):
cg.add(var.set_lp_mode(bool(config[CONF_LOW_POWER_MODE])))
def i2c_device_schema(default_address):
def i2c_device_schema(default_address: int | None) -> cv.Schema:
"""Create a schema for a i2c device.
:param default_address: The default address of the i2c device, can be None to represent
@@ -375,7 +377,7 @@ def i2c_device_schema(default_address):
return cv.Schema(schema)
async def register_i2c_device(var, config):
async def register_i2c_device(var: MockObj, config: ConfigType) -> None:
"""Register an i2c device with the given config.
Sets the i2c bus to use and the i2c address.
@@ -390,11 +392,11 @@ async def register_i2c_device(var, config):
def final_validate_device_schema(
name: str,
*,
min_frequency: cv.frequency = None,
max_frequency: cv.frequency = None,
min_timeout: cv.time_period = None,
max_timeout: cv.time_period = None,
):
min_frequency: Any = None,
max_frequency: Any = None,
min_timeout: Any = None,
max_timeout: Any = None,
) -> cv.Schema:
hub_schema = {}
if (min_frequency is not None) and (max_frequency is not None):
hub_schema[cv.Required(CONF_FREQUENCY)] = cv.Range(
+7 -3
View File
@@ -1,7 +1,11 @@
from typing import Any
import esphome.codegen as cg
from esphome.components import display
import esphome.config_validation as cv
from esphome.const import CONF_DATA, CONF_DIMENSIONS, CONF_POSITION
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
CONF_USER_CHARACTERS = "user_characters"
@@ -9,7 +13,7 @@ lcd_base_ns = cg.esphome_ns.namespace("lcd_base")
LCDDisplay = lcd_base_ns.class_("LCDDisplay", cg.PollingComponent)
def validate_lcd_dimensions(value):
def validate_lcd_dimensions(value: Any) -> list[int]:
value = cv.dimensions(value)
if value[0] > 0x40:
raise cv.Invalid("LCD displays can't have more than 64 columns")
@@ -18,7 +22,7 @@ def validate_lcd_dimensions(value):
return value
def validate_user_characters(value):
def validate_user_characters(value: list[ConfigType]) -> list[ConfigType]:
positions = set()
for conf in value:
if conf[CONF_POSITION] in positions:
@@ -51,7 +55,7 @@ LCD_SCHEMA = display.BASIC_DISPLAY_SCHEMA.extend(
).extend(cv.polling_component_schema("1s"))
async def setup_lcd_display(var, config):
async def setup_lcd_display(var: MockObj, config: ConfigType) -> None:
await display.register_display(var, config)
cg.add(var.set_dimensions(config[CONF_DIMENSIONS][0], config[CONF_DIMENSIONS][1]))
if CONF_USER_CHARACTERS in config:
+1 -1
View File
@@ -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 -1
View File
@@ -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:
+2 -1
View File
@@ -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 (

Some files were not shown because too many files have changed in this diff Show More