pyupgrade -py311-plus

.pre-commit-config.yaml

@@ -4,7 +4,7 @@
 repos:
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.5.4
+    rev: v0.9.2
     hooks:
       # Run the linter.
       - id: ruff

esphome/components/dashboard_import/__init__.py

@@ -1,18 +1,17 @@
 import base64
-import secrets
 from pathlib import Path
-from typing import Optional
 import re
+import secrets
 
 import requests
 from ruamel.yaml import YAML
 
-import esphome.codegen as cg
-import esphome.config_validation as cv
-import esphome.final_validate as fv
 from esphome import git
+import esphome.codegen as cg
 from esphome.components.packages import validate_source_shorthand
-from esphome.const import CONF_REF, CONF_WIFI, CONF_ESPHOME, CONF_PROJECT
+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.yaml_util import dump
 
 dashboard_import_ns = cg.esphome_ns.namespace("dashboard_import")
@@ -84,7 +83,7 @@ async def to_code(config):
 def import_config(
     path: str,
     name: str,
-    friendly_name: Optional[str],
+    friendly_name: str | None,
     project_name: str,
     import_url: str,
     network: str = CONF_WIFI,

esphome/components/esp32/__init__.py

@@ -2,7 +2,6 @@ from dataclasses import dataclass
 import logging
 import os
 from pathlib import Path
-from typing import Optional, Union
 
 from esphome import git
 import esphome.codegen as cg
@@ -142,7 +141,7 @@ class RawSdkconfigValue:
     value: str
 
 
-SdkconfigValueType = Union[bool, int, HexInt, str, RawSdkconfigValue]
+SdkconfigValueType = bool | int | HexInt | str | RawSdkconfigValue
 
 
 def add_idf_sdkconfig_option(name: str, value: SdkconfigValueType):
@@ -159,8 +158,8 @@ def add_idf_component(
     ref: str = None,
     path: str = None,
     refresh: TimePeriod = None,
-    components: Optional[list[str]] = None,
+    components: list[str] | None = None,
-    submodules: Optional[list[str]] = None,
+    submodules: list[str] | None = None,
 ):
     """Add an esp-idf component to the project."""
     if not CORE.using_esp_idf:

esphome/components/libretiny/const.py

@@ -1,5 +1,5 @@
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Callable
 
 import esphome.codegen as cg
 

esphome/components/lvgl/automation.py

@@ -1,4 +1,5 @@
-from typing import Any, Callable
+from collections.abc import Callable
+from typing import Any
 
 from esphome import automation
 import esphome.codegen as cg

esphome/components/lvgl/lv_validation.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 import esphome.codegen as cg
 from esphome.components import image
 from esphome.components.color import CONF_HEX, ColorStruct, from_rgbw
@@ -344,7 +342,7 @@ lv_image_list = LValidator(
 lv_bool = LValidator(cv.boolean, cg.bool_, retmapper=literal)
 
 
-def lv_pct(value: Union[int, float]):
+def lv_pct(value: int | float):
     if isinstance(value, float):
         value = int(value * 100)
     return literal(f"lv_pct({value})")

esphome/components/lvgl/lvcode.py

@@ -1,5 +1,4 @@
 import abc
-from typing import Union
 
 from esphome import codegen as cg
 from esphome.config import Config
@@ -75,7 +74,7 @@ class CodeContext(abc.ABC):
     code_context = None
 
     @abc.abstractmethod
-    def add(self, expression: Union[Expression, Statement]):
+    def add(self, expression: Expression | Statement):
         pass
 
     @staticmethod
@@ -89,13 +88,13 @@ class CodeContext(abc.ABC):
         CodeContext.append(RawStatement("}"))
 
     @staticmethod
-    def append(expression: Union[Expression, Statement]):
+    def append(expression: Expression | Statement):
         if CodeContext.code_context is not None:
             CodeContext.code_context.add(expression)
         return expression
 
     def __init__(self):
-        self.previous: Union[CodeContext | None] = None
+        self.previous: CodeContext | None = None
         self.indent_level = 0
 
     async def __aenter__(self):
@@ -121,7 +120,7 @@ class MainContext(CodeContext):
     Code generation into the main() function
     """
 
-    def add(self, expression: Union[Expression, Statement]):
+    def add(self, expression: Expression | Statement):
         return cg.add(self.indented_statement(expression))
 
 
@@ -144,7 +143,7 @@ class LambdaContext(CodeContext):
         self.capture = capture
         self.where = where
 
-    def add(self, expression: Union[Expression, Statement]):
+    def add(self, expression: Expression | Statement):
         self.code_list.append(self.indented_statement(expression))
         return expression
 
@@ -185,7 +184,7 @@ class LvContext(LambdaContext):
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         await super().__aexit__(exc_type, exc_val, exc_tb)
 
-    def add(self, expression: Union[Expression, Statement]):
+    def add(self, expression: Expression | Statement):
         cg.add(expression)
         return expression
 
@@ -301,7 +300,7 @@ lvgl_static = MockObj("LvglComponent", "::")
 
 
 # equivalent to cg.add() for the current code context
-def lv_add(expression: Union[Expression, Statement]):
+def lv_add(expression: Expression | Statement):
     return CodeContext.append(expression)
 
 

esphome/components/lvgl/widgets/__init__.py

@@ -1,5 +1,5 @@
 import sys
-from typing import Any, Union
+from typing import Any
 
 from esphome import codegen as cg, config_validation as cv
 from esphome.config_validation import Invalid
@@ -263,7 +263,7 @@ async def wait_for_widgets():
     await FakeAwaitable(widgets_wait_generator())
 
 
-async def get_widgets(config: Union[dict, list], id: str = CONF_ID) -> list[Widget]:
+async def get_widgets(config: dict | list, id: str = CONF_ID) -> list[Widget]:
     if not config:
         return []
     if not isinstance(config, list):

esphome/components/opentherm/generate.py

@@ -1,10 +1,11 @@
-from collections.abc import Awaitable
+from collections.abc import Awaitable, Callable
-from typing import Any, Callable, Optional
+from typing import Any
 
 import esphome.codegen as cg
 from esphome.const import CONF_ID
+
 from . import const
-from .schema import TSchema, SettingSchema
+from .schema import SettingSchema, TSchema
 
 opentherm_ns = cg.esphome_ns.namespace("opentherm")
 OpenthermHub = opentherm_ns.class_("OpenthermHub", cg.Component)
@@ -102,7 +103,7 @@ def define_setting_readers(component_type: str, keys: list[str]) -> None:
 
 
 def add_messages(hub: cg.MockObj, keys: list[str], schemas: dict[str, TSchema]):
-    messages: dict[str, tuple[bool, Optional[int]]] = {}
+    messages: dict[str, tuple[bool, int | None]] = {}
     for key in keys:
         messages[schemas[key].message] = (
             schemas[key].keep_updated,
@@ -112,11 +113,10 @@ def add_messages(hub: cg.MockObj, keys: list[str], schemas: dict[str, TSchema]):
         msg_expr = cg.RawExpression(f"esphome::opentherm::MessageId::{msg}")
         if keep_updated:
             cg.add(hub.add_repeating_message(msg_expr))
+        elif order is not None:
+            cg.add(hub.add_initial_message(msg_expr, order))
         else:
-            if order is not None:
+            cg.add(hub.add_initial_message(msg_expr))
-                cg.add(hub.add_initial_message(msg_expr, order))
-            else:
-                cg.add(hub.add_initial_message(msg_expr))
 
 
 def add_property_set(var: cg.MockObj, config_key: str, config: dict[str, Any]) -> None:
@@ -128,7 +128,7 @@ Create = Callable[[dict[str, Any], str, cg.MockObj], Awaitable[cg.Pvariable]]
 
 
 def create_only_conf(
-    create: Callable[[dict[str, Any]], Awaitable[cg.Pvariable]]
+    create: Callable[[dict[str, Any]], Awaitable[cg.Pvariable]],
 ) -> Create:
     return lambda conf, _key, _hub: create(conf)
 

esphome/components/opentherm/schema.py

@@ -2,16 +2,10 @@
 # inputs of the OpenTherm component.
 
 from dataclasses import dataclass
-from typing import Optional, TypeVar, Any
+from typing import Any, TypeVar
 
 import esphome.config_validation as cv
 from esphome.const import (
-    UNIT_CELSIUS,
-    UNIT_EMPTY,
-    UNIT_KILOWATT,
-    UNIT_MICROAMP,
-    UNIT_PERCENT,
-    UNIT_REVOLUTIONS_PER_MINUTE,
     DEVICE_CLASS_COLD,
     DEVICE_CLASS_CURRENT,
     DEVICE_CLASS_EMPTY,
@@ -22,6 +16,12 @@ from esphome.const import (
     STATE_CLASS_MEASUREMENT,
     STATE_CLASS_NONE,
     STATE_CLASS_TOTAL_INCREASING,
+    UNIT_CELSIUS,
+    UNIT_EMPTY,
+    UNIT_KILOWATT,
+    UNIT_MICROAMP,
+    UNIT_PERCENT,
+    UNIT_REVOLUTIONS_PER_MINUTE,
 )
 
 
@@ -61,11 +61,11 @@ TSchema = TypeVar("TSchema", bound=EntitySchema)
 class SensorSchema(EntitySchema):
     accuracy_decimals: int
     state_class: str
-    unit_of_measurement: Optional[str] = None
+    unit_of_measurement: str | None = None
-    icon: Optional[str] = None
+    icon: str | None = None
-    device_class: Optional[str] = None
+    device_class: str | None = None
     disabled_by_default: bool = False
-    order: Optional[int] = None
+    order: int | None = None
 
 
 SENSORS: dict[str, SensorSchema] = {
@@ -461,9 +461,9 @@ SENSORS: dict[str, SensorSchema] = {
 
 @dataclass
 class BinarySensorSchema(EntitySchema):
-    icon: Optional[str] = None
+    icon: str | None = None
-    device_class: Optional[str] = None
+    device_class: str | None = None
-    order: Optional[int] = None
+    order: int | None = None
 
 
 BINARY_SENSORS: dict[str, BinarySensorSchema] = {
@@ -654,7 +654,7 @@ BINARY_SENSORS: dict[str, BinarySensorSchema] = {
 
 @dataclass
 class SwitchSchema(EntitySchema):
-    default_mode: Optional[str] = None
+    default_mode: str | None = None
 
 
 SWITCHES: dict[str, SwitchSchema] = {
@@ -721,9 +721,9 @@ class InputSchema(EntitySchema):
     unit_of_measurement: str
     step: float
     range: tuple[int, int]
-    icon: Optional[str] = None
+    icon: str | None = None
-    auto_max_value: Optional[AutoConfigure] = None
+    auto_max_value: AutoConfigure | None = None
-    auto_min_value: Optional[AutoConfigure] = None
+    auto_min_value: AutoConfigure | None = None
 
 
 INPUTS: dict[str, InputSchema] = {
@@ -834,7 +834,7 @@ class SettingSchema(EntitySchema):
     backing_type: str
     validation_schema: cv.Schema
     default_value: Any
-    order: Optional[int] = None
+    order: int | None = None
 
 
 SETTINGS: dict[str, SettingSchema] = {

esphome/components/opentherm/validate.py

@@ -1,10 +1,10 @@
-from typing import Callable
+from collections.abc import Callable
 
 from voluptuous import Schema
 
 import esphome.config_validation as cv
 
-from . import const, schema, generate
+from . import const, generate, schema
 from .schema import TSchema
 
 

esphome/components/text/__init__.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components import mqtt, web_server
@@ -61,9 +59,9 @@ async def setup_text_core_(
     var,
     config,
     *,
-    min_length: Optional[int],
+    min_length: int | None,
-    max_length: Optional[int],
+    max_length: int | None,
-    pattern: Optional[str],
+    pattern: str | None,
 ):
     await setup_entity(var, config)
 
@@ -90,9 +88,9 @@ async def register_text(
     var,
     config,
     *,
-    min_length: Optional[int] = 0,
+    min_length: int | None = 0,
-    max_length: Optional[int] = 255,
+    max_length: int | None = 255,
-    pattern: Optional[str] = None,
+    pattern: str | None = None,
 ):
     if not CORE.has_id(config[CONF_ID]):
         var = cg.Pvariable(config[CONF_ID], var)
@@ -105,9 +103,9 @@ async def register_text(
 async def new_text(
     config,
     *,
-    min_length: Optional[int] = 0,
+    min_length: int | None = 0,
-    max_length: Optional[int] = 255,
+    max_length: int | None = 255,
-    pattern: Optional[str] = None,
+    pattern: str | None = None,
 ):
     var = cg.new_Pvariable(config[CONF_ID])
     await register_text(

esphome/components/time/__init__.py

@@ -1,6 +1,5 @@
 from importlib import resources
 import logging
-from typing import Optional
 
 import tzlocal
 
@@ -40,7 +39,7 @@ SyncTrigger = time_ns.class_("SyncTrigger", automation.Trigger.template(), cg.Co
 TimeHasTimeCondition = time_ns.class_("TimeHasTimeCondition", Condition)
 
 
-def _load_tzdata(iana_key: str) -> Optional[bytes]:
+def _load_tzdata(iana_key: str) -> bytes | None:
     # From https://tzdata.readthedocs.io/en/latest/#examples
     try:
         package_loc, resource = iana_key.rsplit("/", 1)

esphome/components/uart/__init__.py

@@ -1,36 +1,36 @@
-from typing import Optional
 import re
+
+from esphome import automation, pins
 import esphome.codegen as cg
 import esphome.config_validation as cv
-import esphome.final_validate as fv
-from esphome.yaml_util import make_data_base
-from esphome import pins, automation
 from esphome.const import (
-    CONF_BAUD_RATE,
-    CONF_ID,
-    CONF_NUMBER,
-    CONF_RX_PIN,
-    CONF_TX_PIN,
-    CONF_PORT,
-    CONF_UART_ID,
-    CONF_DATA,
-    CONF_RX_BUFFER_SIZE,
-    CONF_INVERTED,
-    CONF_INVERT,
-    CONF_TRIGGER_ID,
-    CONF_SEQUENCE,
-    CONF_TIMEOUT,
-    CONF_DEBUG,
-    CONF_DIRECTION,
     CONF_AFTER,
+    CONF_BAUD_RATE,
     CONF_BYTES,
+    CONF_DATA,
+    CONF_DEBUG,
     CONF_DELIMITER,
+    CONF_DIRECTION,
     CONF_DUMMY_RECEIVER,
     CONF_DUMMY_RECEIVER_ID,
+    CONF_ID,
+    CONF_INVERT,
+    CONF_INVERTED,
     CONF_LAMBDA,
+    CONF_NUMBER,
+    CONF_PORT,
+    CONF_RX_BUFFER_SIZE,
+    CONF_RX_PIN,
+    CONF_SEQUENCE,
+    CONF_TIMEOUT,
+    CONF_TRIGGER_ID,
+    CONF_TX_PIN,
+    CONF_UART_ID,
     PLATFORM_HOST,
 )
 from esphome.core import CORE
+import esphome.final_validate as fv
+from esphome.yaml_util import make_data_base
 
 CODEOWNERS = ["@esphome/core"]
 uart_ns = cg.esphome_ns.namespace("uart")
@@ -321,12 +321,12 @@ def final_validate_device_schema(
     name: str,
     *,
     uart_bus: str = CONF_UART_ID,
-    baud_rate: Optional[int] = None,
+    baud_rate: int | None = None,
     require_tx: bool = False,
     require_rx: bool = False,
-    data_bits: Optional[int] = None,
+    data_bits: int | None = None,
-    parity: Optional[str] = None,
+    parity: str | None = None,
-    stop_bits: Optional[int] = None,
+    stop_bits: int | None = None,
 ):
     def validate_baud_rate(value):
         if value != baud_rate:

esphome/core/__init__.py

@@ -2,7 +2,7 @@ import logging
 import math
 import os
 import re
-from typing import TYPE_CHECKING, Optional, Union
+from typing import TYPE_CHECKING
 
 from esphome.const import (
     CONF_COMMENT,
@@ -326,7 +326,7 @@ class ID:
         else:
             self.is_manual = is_manual
         self.is_declaration = is_declaration
-        self.type: Optional[MockObjClass] = type
+        self.type: MockObjClass | None = type
 
     def resolve(self, registered_ids):
         from esphome.config_validation import RESERVED_IDS
@@ -477,20 +477,20 @@ class EsphomeCore:
         self.vscode = False
         self.ace = False
         # The name of the node
-        self.name: Optional[str] = None
+        self.name: str | None = None
         # The friendly name of the node
-        self.friendly_name: Optional[str] = None
+        self.friendly_name: str | None = None
         # The area / zone of the node
-        self.area: Optional[str] = None
+        self.area: str | None = None
         # Additional data components can store temporary data in
         # The first key to this dict should always be the integration name
         self.data = {}
         # The relative path to the configuration YAML
-        self.config_path: Optional[str] = None
+        self.config_path: str | None = None
         # The relative path to where all build files are stored
-        self.build_path: Optional[str] = None
+        self.build_path: str | None = None
         # The validated configuration, this is None until the config has been validated
-        self.config: Optional[ConfigType] = None
+        self.config: ConfigType | None = None
         # The pending tasks in the task queue (mostly for C++ generation)
         # This is a priority queue (with heapq)
         # Each item is a tuple of form: (-priority, unique number, task)
@@ -510,7 +510,7 @@ class EsphomeCore:
         # A set of defines to set for the compile process in esphome/core/defines.h
         self.defines: set[Define] = set()
         # A map of all platformio options to apply
-        self.platformio_options: dict[str, Union[str, list[str]]] = {}
+        self.platformio_options: dict[str, str | list[str]] = {}
         # A set of strings of names of loaded integrations, used to find namespace ID conflicts
         self.loaded_integrations = set()
         # A set of component IDs to track what Component subclasses are declared
@@ -545,7 +545,7 @@ class EsphomeCore:
         PIN_SCHEMA_REGISTRY.reset()
 
     @property
-    def address(self) -> Optional[str]:
+    def address(self) -> str | None:
         if self.config is None:
             raise ValueError("Config has not been loaded yet")
 
@@ -558,7 +558,7 @@ class EsphomeCore:
         return None
 
     @property
-    def web_port(self) -> Optional[int]:
+    def web_port(self) -> int | None:
         if self.config is None:
             raise ValueError("Config has not been loaded yet")
 
@@ -571,7 +571,7 @@ class EsphomeCore:
         return None
 
     @property
-    def comment(self) -> Optional[str]:
+    def comment(self) -> str | None:
         if self.config is None:
             raise ValueError("Config has not been loaded yet")
 
@@ -769,7 +769,7 @@ class EsphomeCore:
         _LOGGER.debug("Adding define: %s", define)
         return define
 
-    def add_platformio_option(self, key: str, value: Union[str, list[str]]) -> None:
+    def add_platformio_option(self, key: str, value: str | list[str]) -> None:
         new_val = value
         old_val = self.platformio_options.get(key)
         if isinstance(old_val, list):

esphome/coroutine.py

@@ -43,13 +43,13 @@ the last `yield` expression defines what is returned.
 """
 
 import collections
-from collections.abc import Awaitable, Generator, Iterator
+from collections.abc import Awaitable, Callable, Generator, Iterator
 import functools
 import heapq
 import inspect
 import logging
 import types
-from typing import Any, Callable
+from typing import Any
 
 _LOGGER = logging.getLogger(__name__)
 

esphome/cpp_generator.py

@@ -1,9 +1,9 @@
 import abc
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 import inspect
 import math
 import re
-from typing import Any, Callable, Optional, Union
+from typing import Any
 
 from esphome.core import (
     CORE,
@@ -35,19 +35,19 @@ class Expression(abc.ABC):
         """
 
 
-SafeExpType = Union[
+SafeExpType = (
-    Expression,
+    Expression
-    bool,
+    | bool
-    str,
+    | str
-    str,
+    | str
-    int,
+    | int
-    float,
+    | float
-    TimePeriod,
+    | TimePeriod
-    type[bool],
+    | type[bool]
-    type[int],
+    | type[int]
-    type[float],
+    | type[float]
-    Sequence[Any],
+    | Sequence[Any]
-]
+)
 
 
 class RawExpression(Expression):
@@ -90,7 +90,7 @@ class VariableDeclarationExpression(Expression):
 class ExpressionList(Expression):
     __slots__ = ("args",)
 
-    def __init__(self, *args: Optional[SafeExpType]):
+    def __init__(self, *args: SafeExpType | None):
         # Remove every None on end
         args = list(args)
         while args and args[-1] is None:
@@ -139,7 +139,7 @@ class CallExpression(Expression):
 class StructInitializer(Expression):
     __slots__ = ("base", "args")
 
-    def __init__(self, base: Expression, *args: tuple[str, Optional[SafeExpType]]):
+    def __init__(self, base: Expression, *args: tuple[str, SafeExpType | None]):
         self.base = base
         # TODO: args is always a Tuple, is this check required?
         if not isinstance(args, OrderedDict):
@@ -197,9 +197,7 @@ class ParameterExpression(Expression):
 class ParameterListExpression(Expression):
     __slots__ = ("parameters",)
 
-    def __init__(
+    def __init__(self, *parameters: ParameterExpression | tuple[SafeExpType, str]):
-        self, *parameters: Union[ParameterExpression, tuple[SafeExpType, str]]
-    ):
         self.parameters = []
         for parameter in parameters:
             if not isinstance(parameter, ParameterExpression):
@@ -362,7 +360,7 @@ def safe_exp(obj: SafeExpType) -> Expression:
         return IntLiteral(int(obj.total_seconds))
     if isinstance(obj, TimePeriodMinutes):
         return IntLiteral(int(obj.total_minutes))
-    if isinstance(obj, (tuple, list)):
+    if isinstance(obj, tuple | list):
         return ArrayInitializer(*[safe_exp(o) for o in obj])
     if obj is bool:
         return bool_
@@ -461,7 +459,7 @@ def static_const_array(id_, rhs) -> "MockObj":
     return obj
 
 
-def statement(expression: Union[Expression, Statement]) -> Statement:
+def statement(expression: Expression | Statement) -> Statement:
     """Convert expression into a statement unless is already a statement."""
     if isinstance(expression, Statement):
         return expression
@@ -506,9 +504,9 @@ def with_local_variable(id_: ID, rhs: SafeExpType, callback: Callable, *args) ->
     """
 
     # throw if the callback is async:
-    assert not inspect.iscoroutinefunction(
+    assert not inspect.iscoroutinefunction(callback), (
-        callback
+        "with_local_variable() callback cannot be async!"
-    ), "with_local_variable() callback cannot be async!"
+    )
 
     CORE.add(RawStatement("{"))  # output opening curly brace
     obj = variable(id_, rhs, None, True)
@@ -579,7 +577,7 @@ def new_Pvariable(id_: ID, *args: SafeExpType) -> Pvariable:
     return Pvariable(id_, rhs)
 
 
-def add(expression: Union[Expression, Statement]):
+def add(expression: Expression | Statement):
     """Add an expression to the codegen section.
 
     After this is called, the given given expression will
@@ -588,12 +586,12 @@ def add(expression: Union[Expression, Statement]):
     CORE.add(expression)
 
 
-def add_global(expression: Union[SafeExpType, Statement]):
+def add_global(expression: SafeExpType | Statement):
     """Add an expression to the codegen global storage (above setup())."""
     CORE.add_global(expression)
 
 
-def add_library(name: str, version: Optional[str], repository: Optional[str] = None):
+def add_library(name: str, version: str | None, repository: str | None = None):
     """Add a library to the codegen library storage.
 
     :param name: The name of the library (for example 'AsyncTCP')
@@ -619,7 +617,7 @@ def add_define(name: str, value: SafeExpType = None):
         CORE.add_define(Define(name, safe_exp(value)))
 
 
-def add_platformio_option(key: str, value: Union[str, list[str]]):
+def add_platformio_option(key: str, value: str | list[str]):
     CORE.add_platformio_option(key, value)
 
 
@@ -654,7 +652,7 @@ async def process_lambda(
     parameters: list[tuple[SafeExpType, str]],
     capture: str = "=",
     return_type: SafeExpType = None,
-) -> Union[LambdaExpression, None]:
+) -> LambdaExpression | None:
     """Process the given lambda value into a LambdaExpression.
 
     This is a coroutine because lambdas can depend on other IDs,
@@ -711,8 +709,8 @@ def is_template(value):
 async def templatable(
     value: Any,
     args: list[tuple[SafeExpType, str]],
-    output_type: Optional[SafeExpType],
+    output_type: SafeExpType | None,
-    to_exp: Union[Callable, dict] = None,
+    to_exp: Callable | dict = None,
 ):
     """Generate code for a templatable config option.
 
@@ -817,7 +815,7 @@ class MockObj(Expression):
         assert self.op == "::"
         return MockObj(f"using namespace {self.base}")
 
-    def __getitem__(self, item: Union[str, Expression]) -> "MockObj":
+    def __getitem__(self, item: str | Expression) -> "MockObj":
         next_op = "."
         if isinstance(item, str) and item.startswith("P"):
             item = item[1:]

esphome/dashboard/core.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Coroutine
+from collections.abc import Callable, Coroutine
 import contextlib
 from dataclasses import dataclass
 from functools import partial
@@ -9,7 +9,7 @@ import json
 import logging
 from pathlib import Path
 import threading
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
 
 from esphome.storage_json import ignored_devices_storage_path
 

esphome/dashboard/dns.py

@@ -1,22 +1,16 @@
 from __future__ import annotations
 
-import asyncio
+from asyncio import timeout as async_timeout
-import sys
 
 from icmplib import NameLookupError, async_resolve
 
-if sys.version_info >= (3, 11):
-    from asyncio import timeout as async_timeout
-else:
-    from async_timeout import timeout as async_timeout
-
 
 async def _async_resolve_wrapper(hostname: str) -> list[str] | Exception:
     """Wrap the icmplib async_resolve function."""
     try:
         async with async_timeout(2):
             return await async_resolve(hostname)
-    except (asyncio.TimeoutError, NameLookupError, UnicodeError) as ex:
+    except (TimeoutError, NameLookupError, UnicodeError) as ex:
         return ex
 
 

esphome/dashboard/web_server.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import base64
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 import datetime
 import functools
 import gzip
@@ -17,7 +17,7 @@ import shutil
 import subprocess
 import threading
 import time
-from typing import TYPE_CHECKING, Any, Callable, TypeVar
+from typing import TYPE_CHECKING, Any, TypeVar
 from urllib.parse import urlparse
 
 import tornado

esphome/git.py

@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime
 import hashlib
@@ -5,7 +6,6 @@ import logging
 from pathlib import Path
 import re
 import subprocess
-from typing import Callable, Optional
 import urllib.parse
 
 import esphome.config_validation as cv
@@ -45,12 +45,12 @@ def clone_or_update(
     *,
     url: str,
     ref: str = None,
-    refresh: Optional[TimePeriodSeconds],
+    refresh: TimePeriodSeconds | None,
     domain: str,
     username: str = None,
     password: str = None,
-    submodules: Optional[list[str]] = None,
+    submodules: list[str] | None = None,
-) -> tuple[Path, Optional[Callable[[], None]]]:
+) -> tuple[Path, Callable[[], None] | None]:
     key = f"{url}@{ref}"
 
     if username is not None and password is not None:

esphome/helpers.py

@@ -7,7 +7,6 @@ from pathlib import Path
 import platform
 import re
 import tempfile
-from typing import Union
 from urllib.parse import urlparse
 
 _LOGGER = logging.getLogger(__name__)
@@ -243,7 +242,7 @@ def read_file(path):
         raise EsphomeError(f"Error reading file {path}: {err}") from err
 
 
-def _write_file(path: Union[Path, str], text: Union[str, bytes]):
+def _write_file(path: Path | str, text: str | bytes):
     """Atomically writes `text` to the given path.
 
     Automatically creates all parent directories.
@@ -276,7 +275,7 @@ def _write_file(path: Union[Path, str], text: Union[str, bytes]):
                 _LOGGER.error("Write file cleanup failed: %s", err)
 
 
-def write_file(path: Union[Path, str], text: str):
+def write_file(path: Path | str, text: str):
     try:
         _write_file(path, text)
     except OSError as err:
@@ -285,7 +284,7 @@ def write_file(path: Union[Path, str], text: str):
         raise EsphomeError(f"Could not write file at {path}") from err
 
 
-def write_file_if_changed(path: Union[Path, str], text: str) -> bool:
+def write_file_if_changed(path: Path | str, text: str) -> bool:
     """Write text to the given path, but not if the contents match already.
 
     Returns true if the file was changed.

esphome/loader.py

@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
 import importlib
@@ -8,7 +9,7 @@ import logging
 from pathlib import Path
 import sys
 from types import ModuleType
-from typing import Any, Callable, Optional
+from typing import Any
 
 from esphome.const import SOURCE_FILE_EXTENSIONS
 from esphome.core import CORE
@@ -53,7 +54,7 @@ class ComponentManifest:
         return getattr(self.module, "IS_PLATFORM_COMPONENT", False)
 
     @property
-    def config_schema(self) -> Optional[Any]:
+    def config_schema(self) -> Any | None:
         return getattr(self.module, "CONFIG_SCHEMA", None)
 
     @property
@@ -65,7 +66,7 @@ class ComponentManifest:
         return getattr(self.module, "MULTI_CONF_NO_DEFAULT", False)
 
     @property
-    def to_code(self) -> Optional[Callable[[Any], None]]:
+    def to_code(self) -> Callable[[Any], None] | None:
         return getattr(self.module, "to_code", None)
 
     @property
@@ -88,7 +89,7 @@ class ComponentManifest:
         return getattr(self.module, "CODEOWNERS", [])
 
     @property
-    def final_validate_schema(self) -> Optional[Callable[[ConfigType], None]]:
+    def final_validate_schema(self) -> Callable[[ConfigType], None] | None:
         """Components can declare a `FINAL_VALIDATE_SCHEMA` cv.Schema that gets called
         after the main validation. In that function checks across components can be made.
 
@@ -121,7 +122,7 @@ class ComponentManifest:
 
 class ComponentMetaFinder(importlib.abc.MetaPathFinder):
     def __init__(
-        self, components_path: Path, allowed_components: Optional[list[str]] = None
+        self, components_path: Path, allowed_components: list[str] | None = None
     ) -> None:
         self._allowed_components = allowed_components
         self._finders = []
@@ -132,7 +133,7 @@ class ComponentMetaFinder(importlib.abc.MetaPathFinder):
                 continue
             self._finders.append(finder)
 
-    def find_spec(self, fullname: str, path: Optional[list[str]], target=None):
+    def find_spec(self, fullname: str, path: list[str] | None, target=None):
         if not fullname.startswith("esphome.components."):
             return None
         parts = fullname.split(".")
@@ -159,7 +160,7 @@ def clear_component_meta_finders():
 
 
 def install_meta_finder(
-    components_path: Path, allowed_components: Optional[list[str]] = None
+    components_path: Path, allowed_components: list[str] | None = None
 ):
     sys.meta_path.insert(0, ComponentMetaFinder(components_path, allowed_components))
 

esphome/platformio_api.py

@@ -5,7 +5,6 @@ import os
 from pathlib import Path
 import re
 import subprocess
-from typing import Union
 
 from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, KEY_CORE
 from esphome.core import CORE, EsphomeError
@@ -73,7 +72,7 @@ FILTER_PLATFORMIO_LINES = [
 ]
 
 
-def run_platformio_cli(*args, **kwargs) -> Union[str, int]:
+def run_platformio_cli(*args, **kwargs) -> str | int:
     os.environ["PLATFORMIO_FORCE_COLOR"] = "true"
     os.environ["PLATFORMIO_BUILD_DIR"] = os.path.abspath(CORE.relative_pioenvs_path())
     os.environ.setdefault(
@@ -93,7 +92,7 @@ def run_platformio_cli(*args, **kwargs) -> Union[str, int]:
     return run_external_command(platformio.__main__.main, *cmd, **kwargs)
 
 
-def run_platformio_cli_run(config, verbose, *args, **kwargs) -> Union[str, int]:
+def run_platformio_cli_run(config, verbose, *args, **kwargs) -> str | int:
     command = ["run", "-d", CORE.build_path]
     if verbose:
         command += ["-v"]

esphome/util.py

@@ -6,7 +6,6 @@ from pathlib import Path
 import re
 import subprocess
 import sys
-from typing import Union
 
 from esphome import const
 
@@ -162,7 +161,7 @@ class RedirectText:
 
 def run_external_command(
     func, *cmd, capture_stdout: bool = False, filter_lines: str = None
-) -> Union[int, str]:
+) -> int | str:
     """
     Run a function from an external package that acts like a main method.
 

esphome/writer.py

@@ -3,7 +3,6 @@ import logging
 import os
 from pathlib import Path
 import re
-from typing import Union
 
 from esphome import loader
 from esphome.config import iter_component_configs, iter_components
@@ -132,7 +131,7 @@ def update_storage_json():
     new.save(path)
 
 
-def format_ini(data: dict[str, Union[str, list[str]]]) -> str:
+def format_ini(data: dict[str, str | list[str]]) -> str:
     content = ""
     for key, value in sorted(data.items()):
         if isinstance(value, list):
@@ -212,9 +211,7 @@ def write_platformio_project():
     write_platformio_ini(content)
 
 
-DEFINES_H_FORMAT = (
+DEFINES_H_FORMAT = ESPHOME_H_FORMAT = """\
-    ESPHOME_H_FORMAT
-) = """\
 #pragma once
 #include "esphome/core/macros.h"
 {}

esphome/zeroconf.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Callable
 from dataclasses import dataclass
 import logging
-from typing import Callable
 
 from zeroconf import IPVersion, ServiceInfo, ServiceStateChange, Zeroconf
 from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf

pyproject.toml

@@ -108,6 +108,7 @@ expected-line-ending-format = "LF"
 
 [tool.ruff]
 required-version = ">=0.5.0"
+target-version = "py311"
 
 [tool.ruff.lint]
 select = [