mirror of
https://github.com/esphome/esphome.git
synced 2026-08-31 10:06:03 +00:00
476 lines
16 KiB
Python
476 lines
16 KiB
Python
import errno
|
|
import functools
|
|
from importlib import resources
|
|
import logging
|
|
|
|
import tzlocal
|
|
|
|
from esphome import automation
|
|
from esphome.automation import Condition
|
|
import esphome.codegen as cg
|
|
from esphome.components.zephyr import zephyr_add_prj_conf
|
|
import esphome.config_validation as cv
|
|
from esphome.const import (
|
|
CONF_AT,
|
|
CONF_CRON,
|
|
CONF_DAYS_OF_MONTH,
|
|
CONF_DAYS_OF_WEEK,
|
|
CONF_HOUR,
|
|
CONF_HOURS,
|
|
CONF_ID,
|
|
CONF_MINUTE,
|
|
CONF_MINUTES,
|
|
CONF_MONTHS,
|
|
CONF_ON_TIME,
|
|
CONF_ON_TIME_SYNC,
|
|
CONF_SECOND,
|
|
CONF_SECONDS,
|
|
CONF_TIMEZONE,
|
|
CONF_TRIGGER_ID,
|
|
PLATFORM_BK72XX,
|
|
PLATFORM_ESP32,
|
|
PLATFORM_ESP8266,
|
|
PLATFORM_HOST,
|
|
PLATFORM_LN882X,
|
|
PLATFORM_RP2040,
|
|
PLATFORM_RTL87XX,
|
|
)
|
|
from esphome.core import CORE, CoroPriority, EsphomeError, coroutine_with_priority
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
CODEOWNERS = ["@esphome/core"]
|
|
IS_PLATFORM_COMPONENT = True
|
|
DOMAIN = "time"
|
|
|
|
time_ns = cg.esphome_ns.namespace("time")
|
|
RealTimeClock = time_ns.class_("RealTimeClock", cg.PollingComponent)
|
|
CronTrigger = time_ns.class_("CronTrigger", automation.Trigger.template(), cg.Component)
|
|
SyncTrigger = time_ns.class_("SyncTrigger", automation.Trigger.template(), cg.Component)
|
|
TimeHasTimeCondition = time_ns.class_("TimeHasTimeCondition", Condition)
|
|
|
|
# C++ types for pre-parsed timezone struct generation
|
|
DSTRuleType_cpp = time_ns.enum("DSTRuleType", is_class=True)
|
|
DSTRule_cpp = time_ns.struct("DSTRule")
|
|
ParsedTimezone_cpp = time_ns.struct("ParsedTimezone")
|
|
|
|
|
|
# Map Python DSTRuleType enum values to C++ enum expressions. Built lazily to
|
|
# avoid importing aioesphomeapi (a heavy import) when the time component is only
|
|
# auto-loaded for its schema and never reaches code generation.
|
|
@functools.cache
|
|
def _dst_rule_type_map() -> dict:
|
|
from aioesphomeapi.posix_tz import DSTRuleType as PyDSTRuleType
|
|
|
|
return {
|
|
PyDSTRuleType.NONE: DSTRuleType_cpp.NONE,
|
|
PyDSTRuleType.MONTH_WEEK_DAY: DSTRuleType_cpp.MONTH_WEEK_DAY,
|
|
PyDSTRuleType.JULIAN_NO_LEAP: DSTRuleType_cpp.JULIAN_NO_LEAP,
|
|
PyDSTRuleType.DAY_OF_YEAR: DSTRuleType_cpp.DAY_OF_YEAR,
|
|
}
|
|
|
|
|
|
def _load_tzdata(iana_key: str) -> bytes | None:
|
|
# From https://tzdata.readthedocs.io/en/latest/#examples
|
|
if not iana_key:
|
|
return None
|
|
try:
|
|
package_loc, resource = iana_key.rsplit("/", 1)
|
|
except ValueError:
|
|
# Handle top-level timezone entries like "UTC", "GMT"
|
|
package = "tzdata.zoneinfo"
|
|
resource = iana_key
|
|
else:
|
|
package = "tzdata.zoneinfo." + package_loc.replace("/", ".")
|
|
|
|
try:
|
|
return (resources.files(package) / resource).read_bytes()
|
|
except (FileNotFoundError, ModuleNotFoundError, IsADirectoryError):
|
|
return None
|
|
except OSError as e:
|
|
# Windows raises EINVAL for paths with NTFS-illegal chars (e.g. '<'/'>'
|
|
# in POSIX TZ strings like "<+08>-8" that validate_tz feeds back here).
|
|
if e.errno == errno.EINVAL:
|
|
return None
|
|
raise
|
|
|
|
|
|
def _extract_tz_string(tzfile: bytes) -> str:
|
|
try:
|
|
return tzfile.split(b"\n")[-2].decode()
|
|
except (IndexError, UnicodeDecodeError):
|
|
_LOGGER.error("Could not determine TZ string. Please report this issue.")
|
|
_LOGGER.exception("tzfile contents: %s", tzfile)
|
|
raise
|
|
|
|
|
|
def detect_tz() -> str | None:
|
|
if CORE.target_platform not in {
|
|
PLATFORM_ESP8266,
|
|
PLATFORM_ESP32,
|
|
PLATFORM_RP2040,
|
|
PLATFORM_BK72XX,
|
|
PLATFORM_RTL87XX,
|
|
PLATFORM_LN882X,
|
|
PLATFORM_HOST,
|
|
}:
|
|
return None
|
|
# Avoids duplicate logger messages when multiple time components are configured
|
|
if cached := CORE.data.setdefault(DOMAIN, {}).get(CONF_TIMEZONE):
|
|
return cached
|
|
iana_key = tzlocal.get_localzone_name()
|
|
if iana_key is None:
|
|
raise EsphomeError(
|
|
"Could not automatically determine timezone, please set timezone manually."
|
|
)
|
|
tzfile = _load_tzdata(iana_key)
|
|
if tzfile is None:
|
|
raise EsphomeError(
|
|
"Could not automatically determine timezone, please set timezone manually."
|
|
)
|
|
ret = _extract_tz_string(tzfile)
|
|
_LOGGER.info("Detected timezone '%s'", iana_key)
|
|
_LOGGER.debug(" -> TZ string %s", ret)
|
|
CORE.data.setdefault(DOMAIN, {})[CONF_TIMEZONE] = ret
|
|
return ret
|
|
|
|
|
|
def _parse_cron_int(value, special_mapping, message):
|
|
special_mapping = special_mapping or {}
|
|
if isinstance(value, str) and value in special_mapping:
|
|
return special_mapping[value]
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
raise cv.Invalid(message.format(value)) from None
|
|
|
|
|
|
def _parse_cron_part(part, min_value, max_value, special_mapping):
|
|
if part in ("*", "?"):
|
|
return set(range(min_value, max_value + 1))
|
|
if "/" in part:
|
|
data = part.split("/")
|
|
if len(data) > 2:
|
|
raise cv.Invalid(
|
|
f"Can't have more than two '/' in one time expression, got {part}"
|
|
)
|
|
offset, repeat = data
|
|
offset_n = min_value
|
|
if offset and offset not in ("*", "?"):
|
|
offset_n = _parse_cron_int(
|
|
offset,
|
|
special_mapping,
|
|
"Offset for '/' time expression must be an integer, got {}",
|
|
)
|
|
|
|
try:
|
|
repeat_n = int(repeat)
|
|
except ValueError:
|
|
raise cv.Invalid(
|
|
f"Repeat for '/' time expression must be an integer, got {repeat}"
|
|
) from None
|
|
return set(range(offset_n, max_value + 1, repeat_n))
|
|
if "-" in part:
|
|
data = part.split("-")
|
|
if len(data) > 2:
|
|
raise cv.Invalid(
|
|
f"Can't have more than two '-' in range time expression '{part}'"
|
|
)
|
|
begin, end = data
|
|
begin_n = _parse_cron_int(
|
|
begin, special_mapping, "Number for time range must be integer, got {}"
|
|
)
|
|
end_n = _parse_cron_int(
|
|
end, special_mapping, "Number for time range must be integer, got {}"
|
|
)
|
|
if end_n < begin_n:
|
|
return set(range(end_n, max_value + 1)) | set(range(min_value, begin_n + 1))
|
|
return set(range(begin_n, end_n + 1))
|
|
|
|
return {
|
|
_parse_cron_int(
|
|
part,
|
|
special_mapping,
|
|
"Number for time expression must be an integer, got {}",
|
|
)
|
|
}
|
|
|
|
|
|
def cron_expression_validator(name, min_value, max_value, special_mapping=None):
|
|
def validator(value):
|
|
if isinstance(value, list):
|
|
for v in value:
|
|
if not isinstance(v, int):
|
|
raise cv.Invalid(
|
|
f"Expected integer for {v} '{name}', got {type(v)}"
|
|
)
|
|
if v < min_value or v > max_value:
|
|
raise cv.Invalid(
|
|
f"{name} {v} is out of range (min={min_value} max={max_value})."
|
|
)
|
|
return sorted(value)
|
|
value = cv.string(value)
|
|
values = set()
|
|
for part in value.split(","):
|
|
values |= _parse_cron_part(part, min_value, max_value, special_mapping)
|
|
return validator(list(values))
|
|
|
|
return validator
|
|
|
|
|
|
validate_cron_seconds = cron_expression_validator("seconds", 0, 60)
|
|
validate_cron_minutes = cron_expression_validator("minutes", 0, 59)
|
|
validate_cron_hours = cron_expression_validator("hours", 0, 23)
|
|
validate_cron_days_of_month = cron_expression_validator("days of month", 1, 31)
|
|
validate_cron_months = cron_expression_validator(
|
|
"months",
|
|
1,
|
|
12,
|
|
{
|
|
"JAN": 1,
|
|
"FEB": 2,
|
|
"MAR": 3,
|
|
"APR": 4,
|
|
"MAY": 5,
|
|
"JUN": 6,
|
|
"JUL": 7,
|
|
"AUG": 8,
|
|
"SEP": 9,
|
|
"OCT": 10,
|
|
"NOV": 11,
|
|
"DEC": 12,
|
|
},
|
|
)
|
|
validate_cron_days_of_week = cron_expression_validator(
|
|
"days of week",
|
|
1,
|
|
7,
|
|
{"SUN": 1, "MON": 2, "TUE": 3, "WED": 4, "THU": 5, "FRI": 6, "SAT": 7},
|
|
)
|
|
CRON_KEYS = [
|
|
CONF_SECONDS,
|
|
CONF_MINUTES,
|
|
CONF_HOURS,
|
|
CONF_DAYS_OF_MONTH,
|
|
CONF_MONTHS,
|
|
CONF_DAYS_OF_WEEK,
|
|
]
|
|
|
|
|
|
def validate_cron_raw(value):
|
|
value = cv.string(value)
|
|
value = value.split(" ")
|
|
if len(value) != 6:
|
|
raise cv.Invalid(
|
|
f"Cron expression must consist of exactly 6 space-separated parts, not {len(value)}"
|
|
)
|
|
seconds, minutes, hours, days_of_month, months, days_of_week = value
|
|
return {
|
|
CONF_SECONDS: validate_cron_seconds(seconds),
|
|
CONF_MINUTES: validate_cron_minutes(minutes),
|
|
CONF_HOURS: validate_cron_hours(hours),
|
|
CONF_DAYS_OF_MONTH: validate_cron_days_of_month(days_of_month),
|
|
CONF_MONTHS: validate_cron_months(months),
|
|
CONF_DAYS_OF_WEEK: validate_cron_days_of_week(days_of_week),
|
|
}
|
|
|
|
|
|
def validate_time_at(value):
|
|
value = cv.time_of_day(value)
|
|
return {
|
|
CONF_HOURS: [value[CONF_HOUR]],
|
|
CONF_MINUTES: [value[CONF_MINUTE]],
|
|
CONF_SECONDS: [value[CONF_SECOND]],
|
|
CONF_DAYS_OF_MONTH: validate_cron_days_of_month("*"),
|
|
CONF_MONTHS: validate_cron_months("*"),
|
|
CONF_DAYS_OF_WEEK: validate_cron_days_of_week("*"),
|
|
}
|
|
|
|
|
|
def validate_cron_keys(value):
|
|
if CONF_CRON in value:
|
|
for key in value:
|
|
if key in CRON_KEYS:
|
|
raise cv.Invalid(f"Cannot use option {key} when cron: is specified.")
|
|
if CONF_AT in value:
|
|
raise cv.Invalid("Cannot use option at with cron!")
|
|
cron_ = value[CONF_CRON]
|
|
value = {x: value[x] for x in value if x != CONF_CRON}
|
|
value.update(cron_)
|
|
return value
|
|
if CONF_AT in value:
|
|
for key in value:
|
|
if key in CRON_KEYS:
|
|
raise cv.Invalid(f"Cannot use option {key} when at: is specified.")
|
|
at_ = value[CONF_AT]
|
|
value = {x: value[x] for x in value if x != CONF_AT}
|
|
value.update(at_)
|
|
return value
|
|
return cv.has_at_least_one_key(*CRON_KEYS)(value)
|
|
|
|
|
|
def validate_tz(value: str) -> str:
|
|
value = cv.string_strict(value)
|
|
|
|
tzfile = _load_tzdata(value)
|
|
if tzfile is not None:
|
|
value = _extract_tz_string(tzfile)
|
|
is_iana = True
|
|
else:
|
|
is_iana = False
|
|
|
|
# Validate that the POSIX TZ string is parseable (skip empty strings)
|
|
if value:
|
|
from aioesphomeapi.posix_tz import parse_posix_tz as parse_posix_tz_python
|
|
|
|
try:
|
|
parse_posix_tz_python(value)
|
|
except ValueError as e:
|
|
if is_iana:
|
|
raise cv.Invalid(f"Invalid POSIX timezone string '{value}': {e}") from e
|
|
raise cv.Invalid(
|
|
f"Invalid POSIX timezone string '{value}': {e}. "
|
|
f"If you meant to use an IANA timezone, check the list of valid "
|
|
f"timezones at "
|
|
f"https://en.wikipedia.org/wiki/List_of_tz_database_time_zones"
|
|
) from e
|
|
|
|
return value
|
|
|
|
|
|
TIME_SCHEMA = cv.Schema(
|
|
{
|
|
cv.Optional(CONF_TIMEZONE): cv.All(
|
|
cv.only_with_framework(["arduino", "esp-idf", "host"]),
|
|
validate_tz,
|
|
),
|
|
cv.Optional(CONF_ON_TIME): automation.validate_automation(
|
|
{
|
|
cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(CronTrigger),
|
|
cv.Optional(CONF_SECONDS): validate_cron_seconds,
|
|
cv.Optional(CONF_MINUTES): validate_cron_minutes,
|
|
cv.Optional(CONF_HOURS): validate_cron_hours,
|
|
cv.Optional(CONF_DAYS_OF_MONTH): validate_cron_days_of_month,
|
|
cv.Optional(CONF_MONTHS): validate_cron_months,
|
|
cv.Optional(CONF_DAYS_OF_WEEK): validate_cron_days_of_week,
|
|
cv.Optional(CONF_CRON): validate_cron_raw,
|
|
cv.Optional(CONF_AT): validate_time_at,
|
|
},
|
|
validate_cron_keys,
|
|
),
|
|
cv.Optional(CONF_ON_TIME_SYNC): automation.validate_automation(
|
|
{
|
|
cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(SyncTrigger),
|
|
}
|
|
),
|
|
}
|
|
).extend(
|
|
# ``visibility=ADVANCED`` flags the inherited ``update_interval``
|
|
# field for visual editors — the 15min default is correct for
|
|
# essentially every user, so editors should keep it tucked under
|
|
# "advanced" so it doesn't crowd the form. Validation is
|
|
# unaffected; YAML can override as before.
|
|
cv.polling_component_schema("15min", visibility=cv.Visibility.ADVANCED)
|
|
)
|
|
|
|
|
|
def _emit_dst_rule_fields(prefix, rule):
|
|
"""Emit field-by-field assignments for a DSTRule to avoid rodata struct blob."""
|
|
cg.add(cg.RawExpression(f"{prefix}.time_seconds = {rule.time_seconds}"))
|
|
cg.add(cg.RawExpression(f"{prefix}.day = {rule.day}"))
|
|
cg.add(cg.RawExpression(f"{prefix}.type = {_dst_rule_type_map()[rule.type]}"))
|
|
cg.add(cg.RawExpression(f"{prefix}.month = {rule.month}"))
|
|
cg.add(cg.RawExpression(f"{prefix}.week = {rule.week}"))
|
|
cg.add(cg.RawExpression(f"{prefix}.day_of_week = {rule.day_of_week}"))
|
|
|
|
|
|
def _emit_parsed_timezone_fields(parsed):
|
|
"""Emit field-by-field assignments for a local ParsedTimezone, then set_global_tz().
|
|
|
|
Uses individual assignments on a stack variable instead of a struct initializer
|
|
to keep constants as immediate operands in instructions (.irom0.text/flash)
|
|
rather than a const blob in .rodata (which maps to RAM on ESP8266).
|
|
Wrapped in a scope block to allow multiple time platforms in the same build.
|
|
"""
|
|
cg.add(cg.RawStatement("{"))
|
|
cg.add(cg.RawExpression("time::ParsedTimezone tz{}"))
|
|
cg.add(cg.RawExpression(f"tz.std_offset_seconds = {parsed.std_offset_seconds}"))
|
|
cg.add(cg.RawExpression(f"tz.dst_offset_seconds = {parsed.dst_offset_seconds}"))
|
|
_emit_dst_rule_fields("tz.dst_start", parsed.dst_start)
|
|
_emit_dst_rule_fields("tz.dst_end", parsed.dst_end)
|
|
cg.add(time_ns.set_global_tz(cg.RawExpression("tz")))
|
|
cg.add(cg.RawStatement("}"))
|
|
|
|
|
|
async def setup_time_core_(time_var, config):
|
|
timezone = config.get(CONF_TIMEZONE)
|
|
# an empty timezone is treated as disabling timezones completely as before
|
|
if timezone is None:
|
|
timezone = detect_tz()
|
|
if timezone:
|
|
cg.add_define("USE_TIME_TIMEZONE")
|
|
|
|
if CORE.is_host:
|
|
# Host platform needs setenv("TZ")/tzset() for libc compatibility
|
|
cg.add(time_var.set_timezone(timezone))
|
|
else:
|
|
# Embedded: pre-parse at codegen time, emit struct directly
|
|
from aioesphomeapi.posix_tz import parse_posix_tz as parse_posix_tz_python
|
|
|
|
try:
|
|
parsed = parse_posix_tz_python(timezone)
|
|
_emit_parsed_timezone_fields(parsed)
|
|
except ValueError as e:
|
|
raise EsphomeError(f"Invalid timezone: {timezone}") from e
|
|
|
|
for conf in config.get(CONF_ON_TIME, []):
|
|
trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], time_var)
|
|
|
|
seconds = conf.get(CONF_SECONDS, list(range(61)))
|
|
cg.add(trigger.add_seconds(seconds))
|
|
minutes = conf.get(CONF_MINUTES, list(range(60)))
|
|
cg.add(trigger.add_minutes(minutes))
|
|
hours = conf.get(CONF_HOURS, list(range(24)))
|
|
cg.add(trigger.add_hours(hours))
|
|
days_of_month = conf.get(CONF_DAYS_OF_MONTH, list(range(1, 32)))
|
|
cg.add(trigger.add_days_of_month(days_of_month))
|
|
months = conf.get(CONF_MONTHS, list(range(1, 13)))
|
|
cg.add(trigger.add_months(months))
|
|
days_of_week = conf.get(CONF_DAYS_OF_WEEK, list(range(1, 8)))
|
|
cg.add(trigger.add_days_of_week(days_of_week))
|
|
|
|
await cg.register_component(trigger, conf)
|
|
await automation.build_automation(trigger, [], conf)
|
|
|
|
for conf in config.get(CONF_ON_TIME_SYNC, []):
|
|
trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], time_var)
|
|
|
|
await cg.register_component(trigger, conf)
|
|
await automation.build_automation(trigger, [], conf)
|
|
|
|
|
|
async def register_time(time_var, config):
|
|
await setup_time_core_(time_var, config)
|
|
|
|
|
|
@coroutine_with_priority(CoroPriority.CORE)
|
|
async def to_code(config):
|
|
if CORE.using_zephyr:
|
|
zephyr_add_prj_conf("POSIX_CLOCK", True)
|
|
cg.add_define("USE_TIME")
|
|
cg.add_global(time_ns.using)
|
|
|
|
|
|
@automation.register_condition(
|
|
"time.has_time",
|
|
TimeHasTimeCondition,
|
|
cv.Schema(
|
|
{
|
|
cv.GenerateID(): cv.use_id(RealTimeClock),
|
|
}
|
|
),
|
|
)
|
|
async def time_has_time_to_code(config, condition_id, template_arg, args):
|
|
paren = await cg.get_variable(config[CONF_ID])
|
|
return cg.new_Pvariable(condition_id, template_arg, paren)
|