mirror of
https://github.com/home-assistant/core.git
synced 2025-07-18 18:57:06 +00:00
Use SignalType to improve typing [core] (#114298)
This commit is contained in:
parent
6313571fbc
commit
911a31f860
@ -14,6 +14,7 @@ from homeassistant.helpers.deprecation import (
|
|||||||
check_if_deprecated_constant,
|
check_if_deprecated_constant,
|
||||||
dir_with_deprecated_constants,
|
dir_with_deprecated_constants,
|
||||||
)
|
)
|
||||||
|
from homeassistant.util.signal_type import SignalType
|
||||||
|
|
||||||
LOGGER: Final = logging.getLogger(__package__)
|
LOGGER: Final = logging.getLogger(__package__)
|
||||||
|
|
||||||
@ -68,7 +69,9 @@ ATTR_SOURCE_TYPE: Final = "source_type"
|
|||||||
ATTR_CONSIDER_HOME: Final = "consider_home"
|
ATTR_CONSIDER_HOME: Final = "consider_home"
|
||||||
ATTR_IP: Final = "ip"
|
ATTR_IP: Final = "ip"
|
||||||
|
|
||||||
CONNECTED_DEVICE_REGISTERED: Final = "device_tracker_connected_device_registered"
|
CONNECTED_DEVICE_REGISTERED = SignalType[dict[str, str | None]](
|
||||||
|
"device_tracker_connected_device_registered"
|
||||||
|
)
|
||||||
|
|
||||||
# These can be removed if no deprecated constant are in this module anymore
|
# These can be removed if no deprecated constant are in this module anymore
|
||||||
__getattr__ = partial(check_if_deprecated_constant, module_globals=globals())
|
__getattr__ = partial(check_if_deprecated_constant, module_globals=globals())
|
||||||
|
@ -21,6 +21,7 @@ from homeassistant.helpers.dispatcher import (
|
|||||||
from homeassistant.helpers.typing import ConfigType
|
from homeassistant.helpers.typing import ConfigType
|
||||||
from homeassistant.loader import bind_hass
|
from homeassistant.loader import bind_hass
|
||||||
import homeassistant.util.dt as dt_util
|
import homeassistant.util.dt as dt_util
|
||||||
|
from homeassistant.util.signal_type import SignalType
|
||||||
from homeassistant.util.uuid import random_uuid_hex
|
from homeassistant.util.uuid import random_uuid_hex
|
||||||
|
|
||||||
DOMAIN = "persistent_notification"
|
DOMAIN = "persistent_notification"
|
||||||
@ -50,7 +51,9 @@ class UpdateType(StrEnum):
|
|||||||
UPDATED = "updated"
|
UPDATED = "updated"
|
||||||
|
|
||||||
|
|
||||||
SIGNAL_PERSISTENT_NOTIFICATIONS_UPDATED = "persistent_notifications_updated"
|
SIGNAL_PERSISTENT_NOTIFICATIONS_UPDATED = SignalType[
|
||||||
|
UpdateType, dict[str, Notification]
|
||||||
|
]("persistent_notifications_updated")
|
||||||
|
|
||||||
SCHEMA_SERVICE_NOTIFICATION = vol.Schema(
|
SCHEMA_SERVICE_NOTIFICATION = vol.Schema(
|
||||||
{vol.Required(ATTR_NOTIFICATION_ID): cv.string}
|
{vol.Required(ATTR_NOTIFICATION_ID): cv.string}
|
||||||
|
@ -47,7 +47,7 @@ from .exceptions import (
|
|||||||
)
|
)
|
||||||
from .helpers import device_registry, entity_registry, issue_registry as ir, storage
|
from .helpers import device_registry, entity_registry, issue_registry as ir, storage
|
||||||
from .helpers.debounce import Debouncer
|
from .helpers.debounce import Debouncer
|
||||||
from .helpers.dispatcher import async_dispatcher_send
|
from .helpers.dispatcher import SignalType, async_dispatcher_send
|
||||||
from .helpers.event import (
|
from .helpers.event import (
|
||||||
RANDOM_MICROSECOND_MAX,
|
RANDOM_MICROSECOND_MAX,
|
||||||
RANDOM_MICROSECOND_MIN,
|
RANDOM_MICROSECOND_MIN,
|
||||||
@ -189,7 +189,9 @@ RECONFIGURE_NOTIFICATION_ID = "config_entry_reconfigure"
|
|||||||
|
|
||||||
EVENT_FLOW_DISCOVERED = "config_entry_discovered"
|
EVENT_FLOW_DISCOVERED = "config_entry_discovered"
|
||||||
|
|
||||||
SIGNAL_CONFIG_ENTRY_CHANGED = "config_entry_changed"
|
SIGNAL_CONFIG_ENTRY_CHANGED = SignalType["ConfigEntryChange", "ConfigEntry"](
|
||||||
|
"config_entry_changed"
|
||||||
|
)
|
||||||
|
|
||||||
NO_RESET_TRIES_STATES = {
|
NO_RESET_TRIES_STATES = {
|
||||||
ConfigEntryState.SETUP_RETRY,
|
ConfigEntryState.SETUP_RETRY,
|
||||||
|
@ -81,10 +81,11 @@ from homeassistant.core import (
|
|||||||
from homeassistant.util import slugify
|
from homeassistant.util import slugify
|
||||||
from homeassistant.util.async_ import create_eager_task
|
from homeassistant.util.async_ import create_eager_task
|
||||||
from homeassistant.util.dt import utcnow
|
from homeassistant.util.dt import utcnow
|
||||||
|
from homeassistant.util.signal_type import SignalType, SignalTypeFormat
|
||||||
|
|
||||||
from . import condition, config_validation as cv, service, template
|
from . import condition, config_validation as cv, service, template
|
||||||
from .condition import ConditionCheckerType, trace_condition_function
|
from .condition import ConditionCheckerType, trace_condition_function
|
||||||
from .dispatcher import SignalType, async_dispatcher_connect, async_dispatcher_send
|
from .dispatcher import async_dispatcher_connect, async_dispatcher_send
|
||||||
from .event import async_call_later, async_track_template
|
from .event import async_call_later, async_track_template
|
||||||
from .script_variables import ScriptVariables
|
from .script_variables import ScriptVariables
|
||||||
from .trace import (
|
from .trace import (
|
||||||
@ -155,7 +156,9 @@ _SHUTDOWN_MAX_WAIT = 60
|
|||||||
ACTION_TRACE_NODE_MAX_LEN = 20 # Max length of a trace node for repeated actions
|
ACTION_TRACE_NODE_MAX_LEN = 20 # Max length of a trace node for repeated actions
|
||||||
|
|
||||||
SCRIPT_BREAKPOINT_HIT = SignalType[str, str, str]("script_breakpoint_hit")
|
SCRIPT_BREAKPOINT_HIT = SignalType[str, str, str]("script_breakpoint_hit")
|
||||||
SCRIPT_DEBUG_CONTINUE_STOP = "script_debug_continue_stop_{}_{}"
|
SCRIPT_DEBUG_CONTINUE_STOP: SignalTypeFormat[Literal["continue", "stop"]] = (
|
||||||
|
SignalTypeFormat("script_debug_continue_stop_{}_{}")
|
||||||
|
)
|
||||||
SCRIPT_DEBUG_CONTINUE_ALL = "script_debug_continue_all"
|
SCRIPT_DEBUG_CONTINUE_ALL = "script_debug_continue_all"
|
||||||
|
|
||||||
script_stack_cv: ContextVar[list[int] | None] = ContextVar("script_stack", default=None)
|
script_stack_cv: ContextVar[list[int] | None] = ContextVar("script_stack", default=None)
|
||||||
@ -216,7 +219,9 @@ async def trace_action(
|
|||||||
done = hass.loop.create_future()
|
done = hass.loop.create_future()
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_continue_stop(command=None):
|
def async_continue_stop(
|
||||||
|
command: Literal["continue", "stop"] | None = None,
|
||||||
|
) -> None:
|
||||||
if command == "stop":
|
if command == "stop":
|
||||||
_set_result_unless_done(stop)
|
_set_result_unless_done(stop)
|
||||||
_set_result_unless_done(done)
|
_set_result_unless_done(done)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user